我通过Atomikos和JMS使用Spring Integration和JTA支持绑定到入站和出站的不同Webshpere MQ . 流程如下:
-
JMS入站通道适配器接收消息
-
一些转变
-
JMS出站通道适配器到输出队列
-
发生错误时,
errorChannel
收到该消息 -
异常类型路由器将未处理的错误路由到自定义重新抛出服务,并将其处理到收件人列表路由器,将它们发送到2个错误队列
我的问题是我希望事务提交,即使消息在errorChannel下游(在处理的异常情况和错误队列上) . 据我所知,只有在重新抛出Exception时才会发生回滚(这就是我重新抛出未处理的事件的原因),但在我的情况下,一旦消息到达errorChannel(在它被路由到别处之前),事务就会回滚 .
我究竟做错了什么?
配置如下 .
<jms:inbound-channel-adapter id="jms-in"
destination="input-queue"
connection-factory="inConnectionFactory"
channel="edi-inbound"
acknowledge="transacted">
<poller max-messages-per-poll="${process.jms.inbound.poll.messages-per-poll:1}"
fixed-rate="${process.jms.inbound.poll.rate-millis:60000}"
>
<transactional timeout="${process.tx.timeout-sec:60}"/>
</poller>
</jms:inbound-channel-adapter>
<channel id="edi-inbound"/>
<chain input-channel="edi-inbound" output-channel="edi-transformation-chain">
<object-to-string-transformer/>
<service-activator ref="inbound" method="service"/>
</chain>
<!-- edifact transformation flow -->
<chain input-channel="edi-transformation-chain" output-channel="outbound-message-compose">
<transformer ref="edi2xml-converter"/>
<transformer ref="xml-mapper"/>
</chain>
<chain input-channel="outbound-message-compose" output-channel="outbound-channel">
<service-activator ref="outbound-message-composer" />
</chain>
<channel id="outbound-channel">
<interceptors>
<beans:ref bean="outbound-interceptor" />
</interceptors>
</channel>
<recipient-list-router input-channel="outbound-channel">
<recipient channel="file-outbound"/>
<recipient channel="queue-outbound"/>
</recipient-list-router>
<channel id="queue-outbound"/>
<jms:outbound-channel-adapter id="jms-out" destination="output-queue" channel="queue-outbound" connection-factory="outConnectionFactory"/>
<channel id="file-outbound"/>
<file:outbound-channel-adapter id="file-outbound"
directory="${output.directory}"
filename-generator-expression="headers['${application.header.key.messageid}'] + '_' + new java.util.Date().getTime() + '.xml'"
delete-source-files="true" />
<!-- notification outbound flow -->
<channel id="errorChannel">
<interceptors>
<wire-tap channel="logger"/>
</interceptors>
</channel>
<logging-channel-adapter id="logger" level="INFO"/>
<exception-type-router input-channel="errorChannel" default-output-channel="unhandled-error-channel">
<mapping exception-type="aero.aice.apidcm.integration.exception.HandledException" channel="error-notification-channel" />
</exception-type-router>
<recipient-list-router input-channel="error-notification-channel">
<recipient channel="queue-outbound-error"/>
<recipient channel="queue-inbound-error"/>
</recipient-list-router>
<chain input-channel="queue-outbound-error">
<service-activator ref="outbound-error-composer" />
<jms:outbound-channel-adapter id="jms-out-error"
destination="error-output-queue"
connection-factory="outConnectionFactory"
session-transacted="true"/>
</chain>
<chain input-channel="queue-inbound-error">
<service-activator ref="error-notif-composer" />
<jms:outbound-channel-adapter id="jms-in-error"
destination="error-input-queue"
connection-factory="outConnectionFactory"
session-transacted="true"/>
</chain>
<channel id="unhandled-error-channel" />
<service-activator ref="exception-rethrow" input-channel="unhandled-error-channel"/>
只是要完成,当错误通道上的tx回滚时,两个错误队列在任何情况下都会收到消息(好像出站适配器不参与事务),而正常流程的tx(没有错误发生时)工作得很好 .
1 回答
那是对的 .
因为你使用
Polling Inbound Channel Adapter
. 它的逻辑如下:TX是
pollingTask
代理的一部分,作为AOPTransactionInterceptor
.errorChannel
是this.taskExecutor
的ErrorHandler
的一部分 . 因此,只有当我们从pollingTask
抛出异常时,我们才能达到errorChannel
. 因为我们在那里有TX,所以它当然会被拉回来 .我的观点是:
Polling Inbound Channel Adapter
中的错误处理过程是在TX之外完成的 .考虑切换到
<int-jms:message-driven-channel-adapter>
.