处理错误时如何保持Spring IntegrationFlow的事务一致性？

考虑以下IntegrationFlow：

@Bean
public IntegrationFlow mongoFlow(MongoTemplate mongoTemplate) {
    return IntegrationFlow
            .from(MongoDb.inboundChannelAdapter(mongoTemplate, "{'status' : 'READY'}")
                .update(Update.update("status", "DONE"))
                .collectionName("test").entityClass(Document.class), 
                    p -> p.poller(pm -> pm.fixedDelay(1000L).transactional()))
            .split()
            .handle((GenericHandler<Document>) (p, h) -> {
                if(p.get("message").toString().contains("james"))
                    throw new IllegalArgumentException("no talking to james allowed");
                // process message
                return p;
            }).nullChannel();
}

该流程会锁定记录并将其状态设为DONE，流程成功则提交事务，失败则回滚。但包含“james”的消息会反复失败，因此尝试了以下方案：

方案一：使用错误通道 #

@Bean
public IntegrationFlow mongoFlow(MongoTemplate mongoTemplate) {
    return IntegrationFlow
            .from(MongoDb.inboundChannelAdapter(mongoTemplate, "{'status' : 'READY'}")
                .update(Update.update("status", "DONE"))
                .collectionName("test").entityClass(Document.class), 
                    p -> p.poller(pm -> pm.fixedDelay(1000L).transactional()))
            .enrichHeaders(h -> h.errorChannel("testError"))
            .split()
            .handle((GenericHandler<Document>) (p, h) -> {
                if(p.get("message").toString().contains("james"))
                    throw new IllegalArgumentException("no talking to james allowed");
                // process message
                return p;
            }).nullChannel();
}

@Bean
public IntegrationFlow errorFlow(MongoTemplate mongoTemplate) {
    return IntegrationFlow.from("testError")
            .transform(MessageHandlingException::getFailedMessage)
            .handle((GenericHandler<Document>)(p, h) -> 
                p.append("fails", p.getInteger("fails", 0) + 1)
                    .append("status", p.getInteger("fails") > 2 ? "FAILED" : "READY"))
            .handle(MongoDb.outboundGateway(mongoTemplate).collectionName("test")
              .entityClass(Document.class)
                .collectionCallback((c, m) -> 
                    c.findOneAndUpdate(Filters.eq("uuid", 
                                           ((Document) m.getPayload()).get("uuid")), 
                        Updates.combine(
                            Updates.set("status", ((Document) m.getPayload()).get("status")),
                            Updates.set("fails",  ((Document) m.getPayload()).get("fails" ))
                        )
                    )
                )
            )
            .nullChannel();
}

此方案可运行，但消息从mongoFlow发送到testError通道时事务会终止，希望在同一事务内更新失败次数或设置状态为FAILED后提交。

方案二：使用ExpressionEvaluatingRequestHandlerAdvice #

@Bean
public ExpressionEvaluatingRequestHandlerAdvice handleErrorAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    advice.setReturnFailureExpressionResult(true);
    advice.setOnSuccessExpression(new FunctionExpression<GenericMessage<Document>> (
        gm -> gm.getPayload().append("status", "DONE")
    ));
    advice.setOnFailureExpression(new FunctionExpression<GenericMessage<Document>> (
        gm -> gm.getPayload().append("fails", p.getInteger("fails", 0) + 1)
                    .append("status", p.getInteger("fails") > 2 ? "FAILED" : "READY")
    ));

    return advice;
}

@Bean
public IntegrationFlow mongoFlow(MongoTemplate mongoTemplate) {
    return IntegrationFlow
            .from(MongoDb.inboundChannelAdapter(mongoTemplate, "{'status' : 'READY'}")
                .update(Update.update("status", "PROCESSING"))
                .collectionName("test").entityClass(Document.class), 
                    p -> p.poller(pm -> pm.fixedDelay(1000L).transactional()))
            .split()
            .handle((GenericHandler<Document>) (p, h) -> {
                if(p.get("message").toString().contains("james"))
                    throw new IllegalArgumentException("no talking to james allowed");
                // process message
                return p;
            }, e -> e.advice(handleErrorAdvice()))
          .handle(MongoDb.outboundGateway(mongoTemplate).collectionName("test")
             .entityClass(Document.class)
                .collectionCallback((c, m) -> 
                    c.findOneAndUpdate(Filters.eq("uuid", 
                                       ((Document) m.getPayload()).get("uuid")), 
                        Updates.combine(
                            Updates.set("status", ((Document) m.getPayload()).get("status")),
                            Updates.set("fails",  ((Document) m.getPayload()).get("fails" ))
                        )
                    )
                )
            )
            .nullChannel();
}

此方案能在同一事务内更新记录，但需要设置中间状态PROCESSING来锁定记录，觉得不够优雅。认为错误通道方式更简洁，但不确定当errorChannel为DirectChannel时，事务边界能否跨两个流程。

最新方案：结合RequestHandlerRetryAdvice与ExpressionEvaluatingRequestHandlerAdvice #

@Bean
public IntegrationFlow mongoFlow(MongoTemplate mongoTemplate) {
    return IntegrationFlow
            .from(MongoDb.inboundChannelAdapter(mongoTemplate, "{'status' : 'READY'}")
                .collectionName("test").entityClass(Document.class)
                    .update(Update.update("status", "PROCESSING")), 
                    p -> p.poller(pm -> pm.fixedDelay(1000L).transactional()))
            .split()
            .handle((GenericHandler<Document>) (p, h) -> {
                if(p.get("message").toString().contains("james"))
                    throw new IllegalArgumentException("no talking to james allowed");
                return p;
            }, e -> e.advice(statusAdvice(), retryAdvice()))
            .log( m -> "\n------AFTER ADVICE------\n" + m.getPayload())
            .handle(MongoDb.outboundGateway(mongoTemplate).collectionName("test").entityClass(Document.class)
                .collectionCallback((c, m) -> 
                    c.findOneAndUpdate(Filters.eq("uuid", ((Document) m.getPayload()).get("uuid")), 
                                        Updates.set("status", ((Document) m.getPayload()).get("status"))
                    )
                )
             )
            .nullChannel();
}

@Bean
public RequestHandlerRetryAdvice retryAdvice() {
    RequestHandlerRetryAdvice advice = new RequestHandlerRetryAdvice();
    advice.setRetryTemplate(RetryTemplate.builder()
        .maxAttempts(3)
        .exponentialBackoff(1000L, 2, 10000)
        .build());
    return advice;
}

@Bean
public ExpressionEvaluatingRequestHandlerAdvice statusAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    advice.setReturnFailureExpressionResult(true);
    advice.setOnSuccessExpression(new FunctionExpression<GenericMessage<Document>>(gm -> {
        System.out.println("============================== SUCCESS ADVICE ============================");
        return gm.getPayload().append("status", "DONE");
    }));
    advice.setOnFailureExpression(new FunctionExpression<GenericMessage<Document>>(gm -> {
        System.out.println("============================== FAILURE ADVICE ============================");
        return gm.getPayload().append("status", "FAILED");
    }));

    return advice;
}

疑问 #

1. 当前的实现方向是否正确？
2. 错误通道方案能否实现跨流程的事务边界？

回答 #

关于实现方向

你当前结合RequestHandlerRetryAdvice与ExpressionEvaluatingRequestHandlerAdvice的方案方向是正确的，它很好地解决了重复失败消息的重试与状态流转问题：

• RequestHandlerRetryAdvice负责控制重试次数与退避策略，避免消息立即被标记为失败，给临时故障留出发起恢复的空间；
• ExpressionEvaluatingRequestHandlerAdvice在重试完成后统一处理成功/失败状态更新，逻辑清晰且能保证在事务内完成状态变更。

不过需要注意几个细节：

• 你当前的statusAdvice中，失败时直接将状态设为FAILED，但忽略了失败次数的累加，建议补充这部分逻辑，和之前方案一样判断失败次数是否超过阈值再决定状态是READY还是FAILED；
• 确保MongoDB的事务配置正确，因为MongoDB需要副本集或分片集群才能支持事务，单节点模式下事务不会生效。

关于错误通道的事务边界问题

错误通道方案无法实现跨流程的事务边界，原因如下：

• 当消息处理抛出异常时，原事务会立即标记为回滚状态，即使错误通道使用DirectChannel（同步调用），错误通道内的操作也会在一个新的事务中执行——Spring的事务传播机制中，当当前事务处于回滚状态时，后续操作不会加入原事务，而是开启新事务；
• 此外，MongoDB的事务是基于会话的，原流程的事务会话在异常抛出后会被关闭，错误通道的操作无法复用该会话，自然无法加入原事务。

如果坚持想用错误通道方案，只能放弃“同一事务内更新状态”的需求，或者将错误处理逻辑合并到主流程中（比如用try-catch包裹处理器逻辑，在catch块中直接更新状态），但这种方式会失去错误通道的解耦优势。

总体来说，你当前的Advice组合方案是更合理的选择，中间状态PROCESSING的设计虽然看起来多一步，但它是保证消息不被重复消费的必要机制——在处理过程中锁定消息，避免其他轮询线程同时获取同一条消息，这是分布式消息处理中的常见模式，并非不优雅。

内容的提问来源于stack exchange，提问作者Paul