Spring Integration - JdbcPollingChannelAdapter 提交而不是回滚处理的异常

Question

我将Spring 4.1.x APIs、Spring Integration 4.1.x APIs 和Spring Integration Java DSL 1.0.x APIs 用于EIP 流，我们使用JdbcPollingChannelAdpater 作为流的入口点来使用来自Oracle 数据库表的消息。

即使我们在JdbcPollingChannelAdapter 的Poller 上配置了ErrorHandler，我们看到会话的Transaction 仍然回滚并且在RuntimeException 被抛出并正确处理时未提交ErrorHandler。

读完这个帖子：Spring Transactions - Prevent rollback after unchecked exceptions (RuntimeException)，我觉得不可能阻止回滚，而是强制提交。它是否正确？而且，如果有办法，在安全处理错误时强制提交而不是回滚的最干净的方法是什么？

当前配置：

IntegrationConfig.java：

@Bean
public MessageSource<Object> jdbcMessageSource() {

    JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
            dataSource,
            "select * from SERVICE_TABLE where rownum <= 10 for update skip locked");
    adapter.setUpdateSql("delete from SERVICE_TABLE where SERVICE_MESSAGE_ID in (:id)");
    adapter.setRowMapper(serviceMessageRowMapper);
    adapter.setMaxRowsPerPoll(1);
    adapter.setUpdatePerRow(true);
    return adapter;
}

@SuppressWarnings("unchecked")
@Bean
public IntegrationFlow inFlow() {

    return IntegrationFlows
            .from(jdbcMessageSource(),
                    c -> {
                        c.poller(Pollers.fixedRate(100)
                                .maxMessagesPerPoll(10)
                                .transactional(transactionManager)
                                .errorHandler(errorHandler));
                    })
                .channel(inProcessCh()).get();
}

ErrorHandler.java

@Component
public class ErrorHandler implements org.springframework.util.ErrorHandler {

    @Autowired
    private PlatformTransactionManager transactionManager;

    private static final Logger logger = LogManager.getLogger();

    @Override
    public void handleError(Throwable t) {

        logger.trace("handling error:{}", t.getMessage(), t);

        // handle error code here...

        // we want to force commit the transaction here?
        TransactionStatus txStatus = transactionManager.getTransaction(null);
        transactionManager.commit(txStatus);
    }
}

--- 已编辑以包含 ExpressionEvaluatingRequestHandlerAdvice Bean ---

@Bean
public Advice expressionEvaluatingRequestHandlerAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice expressionEvaluatingRequestHandlerAdvice = new ExpressionEvaluatingRequestHandlerAdvice();
    expressionEvaluatingRequestHandlerAdvice.setTrapException(true);
    expressionEvaluatingRequestHandlerAdvice.setOnSuccessExpression("payload");
    expressionEvaluatingRequestHandlerAdvice
            .setOnFailureExpression("payload");
    expressionEvaluatingRequestHandlerAdvice.setFailureChannel(errorCh());
    return expressionEvaluatingRequestHandlerAdvice;
}

--- 已编辑以显示虚拟测试消息处理程序 ---

    .handle(Message.class,
                    (m, h) -> {

                        boolean forceTestError = m.getHeaders().get("forceTestError");
                        if (forceTestError) {
                            logger.trace("simulated forced TestException");
                            TestException testException = new TestException(
                                    "forced test exception");
                            throw testException;
                        }

                        logger.trace("simulated successful process");

                        return m;
                    }, e-> e.advice(expressionEvaluatingRequestHandlerAdvice())

--- 已编辑以显示 ExecutorChannelInterceptor 方法 ---

@Bean
public IntegrationFlow inFlow() {

    return IntegrationFlows
            .from(jdbcMessageSource(), c -> {
                c.poller(Pollers.fixedRate(100).maxMessagesPerPoll(10)
                        .transactional(transactionManager));
            })
            .enrichHeaders(h -> h.header("errorChannel", errorCh(), true))
            .channel(
                    MessageChannels.executor("testSyncTaskExecutor",
                            syncTaskExecutor()).interceptor(
                            testExecutorChannelInterceptor()))
            .handle(Message.class, (m, h) -> {
                    boolean forceTestError = m.getHeaders().get("forceTestError");
                    if (forceTestError) {
                        logger.trace("simulated forced TestException");
                        TestException testException = new TestException(
                                "forced test exception");
                        throw testException;
                    }

                    logger.trace("simulated successful process");
            }).channel("nullChannel").get();
}

Answer 1

它不会因为你的ErrorHandler 在 TX 完成后已经工作了。

这里有几行源代码 (AbstractPollingEndpoint.Poller)：

@Override
public void run() {
    taskExecutor.execute(new Runnable() {
        @Override
        public void run() {

           .............
                try {
                        if (!pollingTask.call()) {
                            break;
                        }
                        count++;
                    }
                    catch (Exception e) {
            ....
                    }
                }
            }
        });
    }

地点：

1. ErrorHandler 默认应用于taskExecutor (SyncTaskExecutor)。

2. TransactionInterceptor 为 Aspect 被应用于 pollingTask 周围的代理。

因此 TX 在 pollingTask.call() 周围完成并熄灭。只有在那之后，您的 ErrorHandler 才开始在 taskExecutor.execute() 内部工作。

要解决您的问题，您需要确定哪个下游流程部分对 TX 回调不是那么重要，并在那里设置一些 try...catch 或使用 ExpressionEvaluatingRequestHandlerAdvice 来“消除”RuntimeException。

但正如你所注意到的，我的推理必须在 TX 内完成。