java - MessageDrivenBean 两次使用 JMS 消息

Question

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

正如您在该链接上看到的那样，我使用 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);
    }

}
}

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

connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

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

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

似乎一个线程正在接收消息，以便处理它，与此同时，另一个线程正在做完全相同的事情。

我还更改了 JMS session 的类型，在 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.access$100(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$1.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$1.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.access$100(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$1.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$1.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版本，看看能不能用。我会让你们更新。

Answer 1

我终于找到了解决方案!正如我在对原始帖子的编辑中解释的那样，问题似乎与我的代码处理事件时 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。从现在开始，我不再有任何重复问题。