Spring 集成消息存储在错误处理程序之前回滚

Question

我需要在我的 Spring 集成流程中处理某些错误情况。我的流程是使用消息存储并在轮询器上设置错误通道。我曾想过，如果我在错误处理程序中处理消息，则不会发生回滚，但在执行错误流程之前，messageStore remove (delete) 正在回滚。

这是一个重复我的问题的伪流程。

<int:channel id="rollbackTestInput" >
    <int:queue message-store="messageStore"/>
</int:channel>

<int:bridge input-channel="rollbackTestInput" output-channel="createException" >
    <int:poller fixed-rate="50" 
        error-channel="myErrorChannel">
        <int:transactional />
    </int:poller>
</int:bridge>

<int:transformer input-channel="createException" output-channel="infoLogger"
    expression="T(GarbageToForceException).doesNotExist()" />

<int:channel id="myErrorChannel">
    <int:queue/> 
</int:channel>

<!-- JDBC Message Store -->
<bean id="messageStore" class="org.springframework.integration.jdbc.JdbcMessageStore">
    <property name="dataSource">
        <ref bean="dataSource" />
    </property>
</bean>

<bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager" >
<property name="dataSource" ref="dataSource"/>

此流程将导致无限回滚/轮询循环。如何处理错误而不回滚？

Answer 1

这是正确的行为。轮询任务的 TX 建议周围接受 error-channel 逻辑。

代码如下：

@Override
public void run() {
    taskExecutor.execute(new Runnable() {
        @Override
        public void run() {
            int count = 0;
            while (initialized && (maxMessagesPerPoll <= 0 || count < maxMessagesPerPoll)) {
                try {
                    if (!pollingTask.call()) {
                        break;
                    }
                    count++;
                }
                catch (Exception e) {
                    if (e instanceof RuntimeException) {
                        throw (RuntimeException) e;
                    }
                    else {
                        throw new MessageHandlingException(new ErrorMessage(e), e);
                    }
                }
            }
        }
    });
}

TX Advice 在pollingTask.call() 上，但错误处理是从taskExecutor 完成的：

this.taskExecutor = new ErrorHandlingTaskExecutor(this.taskExecutor, this.errorHandler);

您的error-channel 在errorHandler 上配置为MessagePublishingErrorHandler。

要达到您的要求，您可以尝试在<poller> 上关注synchronization-factory：

<int:transaction-synchronization-factory id="txSyncFactory">
    <int:after-rollback channel="myErrorChannel" />
</int:transaction-synchronization-factory>

或者为您的<int:transformer> 提供<request-handler-advice-chain>：

<int:request-handler-advice-chain>
    <bean class="org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice">
        <property name="onFailureExpression" value="#exception" />
        <property name="failureChannel" value="myErrorChannel" />
        <property name="trapException" value="true" />
    </bean>
</int:request-handler-advice-chain>

Answer 2

我找到了一种不同的解决方案，可以满足我的要求并且侵入性较小。

我可以使用通知链来指定我应该回滚和不应该回滚哪些异常，而不是在轮询器上使用事务性。就我而言，我不想回滚大多数异常。所以我回滚 Throwable 并列出我想要回滚的任何特定异常。

<int:bridge input-channel="rollbackTestInput"
    output-channel="createException">
    <int:poller fixed-rate="50"
        error-channel="myErrorChannel">
        <int:advice-chain>
            <int:ref bean="txAdvice"/>
        </int:advice-chain>
    </int:poller>
</int:bridge>

<tx:advice id="txAdvice" transaction-manager="transactionManager">
    <tx:attributes>
        <tx:method name="*" rollback-for="javax.jms.JMSException"
            no-rollback-for="java.lang.Throwable" />
    </tx:attributes>
</tx:advice>