MessageDrivenBean 两次使用 JMS 消息

MessageDrivenBean consuming JMS message twice

我目前 运行 申请 Jboss。我的应用程序需要使用来自 ActiveMQ 的 JMS 消息(我的 ActiveMQ 是我的 Jboss 的一个模块,我遵循了这个过程:https://developer.jboss.org/wiki/IntegrationOfJBossAS7WithActiveMQ)。

如您所见link,我使用 MessageDrivenBean 来使用队列中的消息:

@MessageDriven(activationConfig = {
        @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
        @ActivationConfigProperty(propertyName = "destination", propertyValue = "Mongo-DB")})
public class MongoConsumer implements MessageListener {

@Override
public void onMessage(Message message) {
    TextMessage textMessage = (TextMessage) message;

    try {
        json = textMessage.getText();
        collectionId = JsonUtils.extract(json, "_collectionId");
        uuid = JsonUtils.extract(json, "_uuid");

        queriesMongoDB.save(collectionId, json);
        LOGGER.info("Insert in mongo : {}", uuid);
    } catch (TechnicalException e) {
        LOGGER.error("Something went wrong while calling Mongo : {}", e);
        this.rollbackMdb(json);
    } catch (JMSException e) {
        LOGGER.error("Something went wrong with the Mongo Consumer", e);
    }

}
}

我的会话当前以自动确认模式启动:

connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

当我在消息很少的情况下使用它时,所有这些都工作得很好。但是,当我对我的应用程序进行基准测试并发送大量消息时,有时,MDB 会使用两次相同的消息,并将其保存在 MongoDB 中,当然还要再两次。

问题是,只有当我有很多消息(比如 200k)时才会发生这种情况,我得到大约 10 个重复项。

似乎有一个线程正在接收消息,以便对其进行处理,与此同时,另一个线程正在做完全相同的事情。

我还更改了 JMS 会话的类型,在 CLIENT_acknowledge 模式下,并添加:

 message.acknowledge();

在我的方法开始时,但这没有帮助。

PS : 抱歉我的英语不好。

编辑:

我只是重现了错误并阅读了 server.log,我得到了重复的此类错误:

16:41:35,376 WARN  [org.apache.activemq.TransactionContext] (default-threads - 39) commit of: XID:[131077,globalId=0:ffff0a48263f:-669b4574:57e296f4:23fc95,branchId=0:f
fff0a48263f:-669b4574:57e296f4:23fc96] failed with: javax.jms.JMSException: Transaction 'XID:[131077,globalId=0:ffff0a48263f:-669b4574:57e296f4:23fc95,branchId=0:ffff0a
48263f:-669b4574:57e296f4:23fc96]' has not been started. xaErrorCode:-4: javax.jms.JMSException: Transaction 'XID:[131077,globalId=0:ffff0a48263f:-669b4574:57e296f4:23f
c95,branchId=0:ffff0a48263f:-669b4574:57e296f4:23fc96]' has not been started. xaErrorCode:-4
        at org.apache.activemq.util.JMSExceptionSupport.create(JMSExceptionSupport.java:54) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.ActiveMQConnection.syncSendPacket(ActiveMQConnection.java:1420) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.TransactionContext.syncSendPacketWithInterruptionHandling(TransactionContext.java:761) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.TransactionContext.commit(TransactionContext.java:562) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.ra.LocalAndXATransaction.commit(LocalAndXATransaction.java:92)
        at com.arjuna.ats.internal.jta.resources.arjunacore.XAResourceRecord.topLevelOnePhaseCommit(XAResourceRecord.java:682)
        at com.arjuna.ats.arjuna.coordinator.BasicAction.onePhaseCommit(BasicAction.java:2278)
        at com.arjuna.ats.arjuna.coordinator.BasicAction.End(BasicAction.java:1479)
        at com.arjuna.ats.arjuna.coordinator.TwoPhaseCoordinator.end(TwoPhaseCoordinator.java:98)
        at com.arjuna.ats.arjuna.AtomicAction.commit(AtomicAction.java:162)
        at com.arjuna.ats.internal.jta.transaction.arjunacore.TransactionImple.commitAndDisassociate(TransactionImple.java:1189)
        at com.arjuna.ats.internal.jta.transaction.arjunacore.BaseTransaction.commit(BaseTransaction.java:126)
        at com.arjuna.ats.jbossatx.BaseTransactionManagerDelegate.commit(BaseTransactionManagerDelegate.java:75)
        at org.jboss.as.ejb3.inflow.MessageEndpointInvocationHandler.afterDelivery(MessageEndpointInvocationHandler.java:72) [jboss-as-ejb3.jar:7.5.7.Final-redhat-3]
        at sun.reflect.GeneratedMethodAccessor67.invoke(Unknown Source) [:1.8.0_66]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) [rt.jar:1.8.0_66]
        at java.lang.reflect.Method.invoke(Method.java:497) [rt.jar:1.8.0_66]
        at org.jboss.as.ejb3.inflow.AbstractInvocationHandler.handle(AbstractInvocationHandler.java:60) [jboss-as-ejb3.jar:7.5.7.Final-redhat-3]
        at org.jboss.as.ejb3.inflow.MessageEndpointInvocationHandler.doInvoke(MessageEndpointInvocationHandler.java:136) [jboss-as-ejb3.jar:7.5.7.Final-redhat-3]
        at org.jboss.as.ejb3.inflow.AbstractInvocationHandler.invoke(AbstractInvocationHandler.java:73) [jboss-as-ejb3.jar:7.5.7.Final-redhat-3]
        at com.sun.proxy.$Proxy94.afterDelivery(Unknown Source)
        at org.apache.activemq.ra.MessageEndpointProxy$MessageEndpointAlive.afterDelivery(MessageEndpointProxy.java:128)
        at org.apache.activemq.ra.MessageEndpointProxy.afterDelivery(MessageEndpointProxy.java:69)
        at org.apache.activemq.ra.ServerSessionImpl.afterDelivery(ServerSessionImpl.java:225)
        at org.apache.activemq.ActiveMQSession.run(ActiveMQSession.java:1016) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.ra.ServerSessionImpl.run(ServerSessionImpl.java:169)
        at org.jboss.jca.core.workmanager.WorkWrapper.run(WorkWrapper.java:215)
        at org.jboss.threads.SimpleDirectExecutor.execute(SimpleDirectExecutor.java:33)
        at org.jboss.threads.QueueExecutor.runTask(QueueExecutor.java:808)
        at org.jboss.threads.QueueExecutor.access0(QueueExecutor.java:45)
        at org.jboss.threads.QueueExecutor$Worker.run(QueueExecutor.java:849)
        at java.lang.Thread.run(Thread.java:745) [rt.jar:1.8.0_66]
        at org.jboss.threads.JBossThread.run(JBossThread.java:122)
Caused by: javax.transaction.xa.XAException: Transaction 'XID:[131077,globalId=0:ffff0a48263f:-669b4574:57e296f4:23fc95,branchId=0:ffff0a48263f:-669b4574:57e296f4:23fc96]' has not been started. xaErrorCode:-4
        at org.apache.activemq.transaction.XATransaction.newXAException(XATransaction.java:174)
        at org.apache.activemq.broker.TransactionBroker.getTransaction(TransactionBroker.java:368)
        at org.apache.activemq.broker.TransactionBroker.commitTransaction(TransactionBroker.java:252)
        at org.apache.activemq.broker.MutableBrokerFilter.commitTransaction(MutableBrokerFilter.java:117)
        at org.apache.activemq.broker.TransportConnection.processCommitTransactionOnePhase(TransportConnection.java:498)
        at org.apache.activemq.command.TransactionInfo.visit(TransactionInfo.java:100) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:334)
        at org.apache.activemq.broker.TransportConnection.onCommand(TransportConnection.java:188)
        at org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:116) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.transport.vm.VMTransport.iterate(VMTransport.java:248)
        at org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:133) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.thread.PooledTaskRunner.run(PooledTaskRunner.java:48) [activemq-client-5.10.0.jar:5.10.0]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [rt.jar:1.8.0_66]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [rt.jar:1.8.0_66]
        at java.lang.Thread.run(Thread.java:745) [rt.jar:1.8.0_66]

16:41:35,410 WARN  [com.arjuna.ats.jta] (default-threads - 39) ARJUNA016039: onePhaseCommit on < formatId=131077, gtrid_length=29, bqual_length=36, tx_uid=0:ffff0a48263f:-669b4574:57e296f4:23fc95, node_name=1, branch_uid=0:ffff0a48263f:-669b4574:57e296f4:23fc96, subordinatenodename=null, eis_name=unknown eis name > ([org.apache.activemq.ra.LocalAndXATransaction@7e5f0986,TransactionContext{transactionId=null,connection=ActiveMQConnection {id=ID:tsfla902v-34440-1474467574382-7:1,clientId=ID:tsfla902v-34440-1474467574382-6:1,started=true}}]) failed with exception XAException.XAER_NOTA: javax.transaction.xa.XAException: Transaction 'XID:[131077,globalId=0:ffff0a48263f:-669b4574:57e296f4:23fc95,branchId=0:ffff0a48263f:-669b4574:57e296f4:23fc96]' has not been started. xaErrorCode:-4
        at org.apache.activemq.TransactionContext.toXAException(TransactionContext.java:786) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.TransactionContext.commit(TransactionContext.java:595) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.ra.LocalAndXATransaction.commit(LocalAndXATransaction.java:92)
        at com.arjuna.ats.internal.jta.resources.arjunacore.XAResourceRecord.topLevelOnePhaseCommit(XAResourceRecord.java:682)
        at com.arjuna.ats.arjuna.coordinator.BasicAction.onePhaseCommit(BasicAction.java:2278)
        at com.arjuna.ats.arjuna.coordinator.BasicAction.End(BasicAction.java:1479)
        at com.arjuna.ats.arjuna.coordinator.TwoPhaseCoordinator.end(TwoPhaseCoordinator.java:98)
        at com.arjuna.ats.arjuna.AtomicAction.commit(AtomicAction.java:162)
        at com.arjuna.ats.internal.jta.transaction.arjunacore.TransactionImple.commitAndDisassociate(TransactionImple.java:1189)
        at com.arjuna.ats.internal.jta.transaction.arjunacore.BaseTransaction.commit(BaseTransaction.java:126)
        at com.arjuna.ats.jbossatx.BaseTransactionManagerDelegate.commit(BaseTransactionManagerDelegate.java:75)
        at org.jboss.as.ejb3.inflow.MessageEndpointInvocationHandler.afterDelivery(MessageEndpointInvocationHandler.java:72) [jboss-as-ejb3.jar:7.5.7.Final-redhat-3]
        at sun.reflect.GeneratedMethodAccessor67.invoke(Unknown Source) [:1.8.0_66]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) [rt.jar:1.8.0_66]
        at java.lang.reflect.Method.invoke(Method.java:497) [rt.jar:1.8.0_66]
        at org.jboss.as.ejb3.inflow.AbstractInvocationHandler.handle(AbstractInvocationHandler.java:60) [jboss-as-ejb3.jar:7.5.7.Final-redhat-3]
        at org.jboss.as.ejb3.inflow.MessageEndpointInvocationHandler.doInvoke(MessageEndpointInvocationHandler.java:136) [jboss-as-ejb3.jar:7.5.7.Final-redhat-3]
        at org.jboss.as.ejb3.inflow.AbstractInvocationHandler.invoke(AbstractInvocationHandler.java:73) [jboss-as-ejb3.jar:7.5.7.Final-redhat-3]
        at com.sun.proxy.$Proxy94.afterDelivery(Unknown Source)
        at org.apache.activemq.ra.MessageEndpointProxy$MessageEndpointAlive.afterDelivery(MessageEndpointProxy.java:128)
        at org.apache.activemq.ra.MessageEndpointProxy.afterDelivery(MessageEndpointProxy.java:69)
        at org.apache.activemq.ra.ServerSessionImpl.afterDelivery(ServerSessionImpl.java:225)
        at org.apache.activemq.ActiveMQSession.run(ActiveMQSession.java:1016) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.ra.ServerSessionImpl.run(ServerSessionImpl.java:169)
        at org.jboss.jca.core.workmanager.WorkWrapper.run(WorkWrapper.java:215)
        at org.jboss.threads.SimpleDirectExecutor.execute(SimpleDirectExecutor.java:33)
        at org.jboss.threads.QueueExecutor.runTask(QueueExecutor.java:808)
        at org.jboss.threads.QueueExecutor.access0(QueueExecutor.java:45)
        at org.jboss.threads.QueueExecutor$Worker.run(QueueExecutor.java:849)
        at java.lang.Thread.run(Thread.java:745) [rt.jar:1.8.0_66]
        at org.jboss.threads.JBossThread.run(JBossThread.java:122)
Caused by: javax.transaction.xa.XAException: Transaction 'XID:[131077,globalId=0:ffff0a48263f:-669b4574:57e296f4:23fc95,branchId=0:ffff0a48263f:-669b4574:57e296f4:23fc96]' has not been started. xaErrorCode:-4
        at org.apache.activemq.transaction.XATransaction.newXAException(XATransaction.java:174)
        at org.apache.activemq.broker.TransactionBroker.getTransaction(TransactionBroker.java:368)
        at org.apache.activemq.broker.TransactionBroker.commitTransaction(TransactionBroker.java:252)
        at org.apache.activemq.broker.MutableBrokerFilter.commitTransaction(MutableBrokerFilter.java:117)
        at org.apache.activemq.broker.TransportConnection.processCommitTransactionOnePhase(TransportConnection.java:498)
        at org.apache.activemq.command.TransactionInfo.visit(TransactionInfo.java:100) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:334)
        at org.apache.activemq.broker.TransportConnection.onCommand(TransportConnection.java:188)
        at org.apache.activemq.transport.ResponseCorrelator.onCommand(ResponseCorrelator.java:116) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.transport.vm.VMTransport.iterate(VMTransport.java:248)
        at org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:133) [activemq-client-5.10.0.jar:5.10.0]
        at org.apache.activemq.thread.PooledTaskRunner.run(PooledTaskRunner.java:48) [activemq-client-5.10.0.jar:5.10.0]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [rt.jar:1.8.0_66]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [rt.jar:1.8.0_66]
        at java.lang.Thread.run(Thread.java:745) [rt.jar:1.8.0_66]

16:41:35,418 WARN  [com.arjuna.ats.arjuna] (default-threads - 39) ARJUNA012084: One-phase commit of action 0:ffff0a48263f:-669b4574:57e296f4:23fc95 received heuristic decision: TwoPhaseOutcome.HEURISTIC_HAZARD

编辑 2:

似乎有人在这里遇到了同样的问题:https://issues.apache.org/jira/browse/AMQ-5953

我ActiveMQ模块的版本是5.11,我会尝试安装5.13版本,看看能不能用。我让你们更新。

我终于找到了解决办法!正如我在对原始 post 的编辑中解释的那样,问题似乎与我的代码处理事件时 XATransaction 被销毁有关。事务在队列中重新创建,之后我的代码正在处理它。

我在 server.log 中得到了这个堆栈:

16:41:35,376 WARN  [org.apache.activemq.TransactionContext] (default-threads - 39) commit of: XID:[131077,globalId=0:ffff0a48263f:-669b4574:57e296f4:23fc95,branchId=0:f
fff0a48263f:-669b4574:57e296f4:23fc96] failed with: javax.jms.JMSException: Transaction 'XID:[131077,globalId=0:ffff0a48263f:-669b4574:57e296f4:23fc95,branchId=0:ffff0a
48263f:-669b4574:57e296f4:23fc96]' has not been started. xaErrorCode:-4: javax.jms.JMSException: Transaction 'XID:[131077,globalId=0:ffff0a48263f:-669b4574:57e296f4:23f
c95,branchId=0:ffff0a48263f:-669b4574:57e296f4:23fc96]' has not been started. xaErrorCode:-4

然后,我发现了这个:https://issues.apache.org/jira/browse/AMQ-5953。我 JBoss 中的 ActiveMQ 模块是 5.11 版,我设法安装了 5.14。从现在开始,我再也没有重复的问题了。