回滚 xa 事务 activemq + spring-integration + spring-jms 时没有重新交付

Question

这是我的第一篇文章，所以希望我不会把格式弄乱。

我有一个 spring 集成路由，它按如下方式路由消息：

网关 -> 路由器 -> 出站 jms 通道 -> 消息驱动通道 -> 服务激活器

我正在使用 spring 集成版本 4.0.3 和 activemq 版本 5.10.0。在这个特定的实例中，我在 weblogic 容器中运行（尽管情况并非总是如此）。

路由正在将 JMS 消息发布到 activemq 队列。如果在侦听器的执行过程中遇到运行时异常，我希望回滚侦听器中登记的任何事务资源并重新传递消息。目前我看到回滚但没有重新传递消息。

我的spring配置如下：

 <!-- TaskExecutor is defined elsewhere and is our own object, it's effectively the equivalent of calling Executors.newCachedThreadPool(). -->

<bean id="TransactionManager" class="org.springframework.transaction.jta.WebLogicJtaTransactionManager"/>

<bean id="XADataSource" class="org.springframework.jndi.JndiObjectFactoryBean">
    <property name="jndiName" value="${xaJndiName}"/>
</bean>

<amq:broker brokerName="messageBroker" useJmx="true" persistent="true" start="true" schedulerSupport="false">
    <amq:transportConnectors>
        <amq:transportConnector id="tcpConnector"  uri="tcp://garethDell.leeds.retailexp.com:9783"/>
    </amq:transportConnectors>
    <amq:persistenceAdapter>
        <amq:jdbcPersistenceAdapter dataSource="#XADataSource"/>
    </amq:persistenceAdapter>
</amq:broker>

<amq:xaConnectionFactory id="amqConnectionFactory" brokerURL="tcp://garethDell.leeds.retailexp.com:9783" useAsyncSend="false">
    <amq:prefetchPolicy>
        <amq:prefetchPolicy all="1"/>
    </amq:prefetchPolicy>
    <amq:redeliveryPolicy>
        <amq:redeliveryPolicy id="redeliveryPolicy" initialRedeliveryDelay="5000" maximumRedeliveries="5" useExponentialBackOff="true" backOffMultiplier="2" queue="*" />
    </amq:redeliveryPolicy>
</amq:xaConnectionFactory>

<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
    <property name="targetConnectionFactory" ref="amqConnectionFactory"/>
    <property name="reconnectOnException" value="true"/>
</bean>

<int:gateway id="messageGateway" service-interface="com.retailexp.amp.server.spring.MessageGateway" default-request-channel="requestChannel"/>
<bean id="MessagePublisher" class="com.retailexp.amp.server.common.publishing.impl.MessagePublisher">
    <property name="gateway" ref="messageGateway"/>
</bean>

<!-- Routes the messages to the appropriate destination, each queue / topic should have an appropriate channel associated with it which must be registered here-->
<int:channel id="requestChannel"/>
<int:header-value-router id="destinationRouter" input-channel="requestChannel" header-name="destination">
    <int:mapping value="test" channel="testOutChannel"/>
</int:header-value-router>

<int:channel id="testOutChannel"/>
<int-jms:outbound-channel-adapter id="testOut" channel="testOutChannel" connection-factory="connectionFactory" destination="testQueue"/>

<int:channel id="testListener"/>
<int-jms:message-driven-channel-adapter id="testIn" connection-factory="connectionFactory" container="testQueueContainer" channel="testListener"/>

<!-- Define the queues -->
<bean id="testQueue" class="org.apache.activemq.command.ActiveMQQueue">
    <constructor-arg value="test"/>
</bean>

<!-- Define the message listeners -->
<bean id="testMessageHandler" class="com.retailexp.amp.server.spring.TestListener"/>
<int:service-activator input-channel="testListener" ref="testMessageHandler" method="onMessage"/>

<bean id="testQueueContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory"/>
    <property name="transactionManager" ref="TransactionManager"/>
    <property name="taskExecutor" ref="TaskExecutor"/>
    <property name="maxConcurrentConsumers" value="10"/>
    <property name="sessionTransacted" value="true"/>
    <property name="sessionAcknowledgeMode" value="0"/>
    <property name="destination" ref="testQueue"/>
    <property name="cacheLevelName" value="CACHE_CONSUMER"/>
</bean>

我的监听器在这个阶段非常直接，它会注销消息然后保存我们的一个域对象，最后抛出运行时异常以强制回滚。此阶段的回滚出现在日志中，并且数据库更改已正确回滚，因此我知道在侦听器中我正在获取事务回滚。我假设消费者没有在应有的时候作为 XA 事务的一部分登记，因此消息的消费是作为单独事务的一部分提交的（在某种程度上是非事务性的？）。

由于 TransactionContext 中的 xa jdbc 驱动程序，我看到设置事务隔离级别时出错，但我认为这不是问题的原因 - 我尝试使用各种不同的隔离级别，例如8 TRANSACTION_SERIALIZABLE 所以我认为这只是驱动程序本身的限制。

例外：

[11 Sep 2014 16:53:07,464] [ActiveMQ Transport: tcp:///192.168.50.100:58292@9783  ] [            ] [            ] TRACE jdbc.TransactionContext                  - Cannot set transaction isolation to 1 due Due to vendor limitations, setting transaction isolation for "Oracle XA" JDBC XA driver is not supported.. This exception is ignored.
java.sql.SQLException: Due to vendor limitations, setting transaction isolation for "Oracle XA" JDBC XA driver is not supported.
    at weblogic.jdbc.wrapper.JTAConnection.setTransactionIsolation(JTAConnection.java:492)
    at org.apache.activemq.store.jdbc.TransactionContext.getConnection(TransactionContext.java:74)
    at org.apache.activemq.store.jdbc.adapter.DefaultJDBCAdapter.doAddMessage(DefaultJDBCAdapter.java:220)
    at org.apache.activemq.store.jdbc.JDBCMessageStore.addMessage(JDBCMessageStore.java:126)
    at org.apache.activemq.store.memory.MemoryTransactionStore.addMessage(MemoryTransactionStore.java:343)
    at org.apache.activemq.store.memory.MemoryTransactionStore$1.asyncAddQueueMessage(MemoryTransactionStore.java:159)
    at org.apache.activemq.broker.region.Queue.doMessageSend(Queue.java:910)
    at org.apache.activemq.broker.region.Queue.send(Queue.java:733)
    at org.apache.activemq.broker.region.AbstractRegion.send(AbstractRegion.java:424)
    at org.apache.activemq.broker.region.RegionBroker.send(RegionBroker.java:445)
    at org.apache.activemq.broker.jmx.ManagedRegionBroker.send(ManagedRegionBroker.java:297)
    at org.apache.activemq.broker.BrokerFilter.send(BrokerFilter.java:147)
    at org.apache.activemq.broker.CompositeDestinationBroker.send(CompositeDestinationBroker.java:96)
    at org.apache.activemq.broker.TransactionBroker.send(TransactionBroker.java:307)
    at org.apache.activemq.broker.MutableBrokerFilter.send(MutableBrokerFilter.java:152)
    at org.apache.activemq.broker.TransportConnection.processMessage(TransportConnection.java:496)
    at org.apache.activemq.command.ActiveMQMessage.visit(ActiveMQMessage.java:756)
    at org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:294)
    at org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:148)
    at org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
    at org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:113)
    at org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(AbstractInactivityMonitor.java:270)
    at org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
    at org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:214)
    at org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:196)
    at java.lang.Thread.run(Thread.java:722)

我有跟踪级别的日志，如果需要，可以使用它们，但它们非常冗长，但让我超过了这篇文章的字符数限制。让我知道他们是否有帮助，我会尽力提供。

谁能看看上面的配置有什么明显的问题吗？

编辑：此处提供跟踪日志：http://pastebin.com/0awmzY2D

谢谢

Answer 1

事实证明，在消息到达侦听器时，会话持有的 TransactionContext 上没有 XID。实际上会话没有被纳入事务，所以我得到了一个新事务，它只是包装了侦听器代码。

我做了一些关于应该在哪里进行登记的问题，看起来 xa 连接工厂在创建事务时没有做任何事情来登记会话，回想起来我不知道为什么我认为它会因为它没有参考事务管理器。

更改为 XAPooledConnectionFactory 会导致 XaConnectionPool 执行自动登记，配置更改如下：

<bean id="connectionFactory" class="org.apache.activemq.jms.pool.XaPooledConnectionFactory">
    <property name="connectionFactory" ref="xaConnectionFactory"/>
    <property name="tmJndiName" value="weblogic.transaction.TransactionManager"/>
    <property name="tmFromJndi" value="true"/>
</bean>

<bean id="testQueueContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="connectionFactory" ref="connectionFactory"/>
    <property name="transactionManager" ref="TransactionManager"/>
    <property name="taskExecutor" ref="TaskExecutor"/>
    <property name="maxConcurrentConsumers" value="10"/>
    <property name="sessionTransacted" value="true"/>
    <property name="destination" ref="testQueue"/>
</bean>

我现在正在重新发送消息。我仍然遇到了一个问题，即未遵守交付延迟和回退政策（尽管奇怪的是 maximumRedeliveries 是），但我确信我可以解决这个问题。

感谢您的宝贵时间。