Spring Integration Scatter-Gather异常事务回滚配置求助

你当前的核心问题分为两个层面：

1. 单个处理器事务未生效：处理器中的数据库操作未被事务上下文包裹，导致抛出异常时无法触发回滚
2. 异步场景下的全局事务一致性：由于使用了带TaskExecutor的PublishSubscribeChannel，多个处理器并行异步执行，各自事务独立。当其中一个处理器失败时，已完成的处理器事务无法回滚（比如successHandler中状态为"bar"的数据会被持久化）

一、实现单个处理器异常回滚 #

先确保每个处理器的数据库操作被事务管理，有两种配置方式：

方式1：直接使用@Transactional注解 #

在处理器Bean上添加事务注解，指定回滚规则（需确保项目已正确配置事务管理器）：

@Bean
@Transactional(rollbackFor = RuntimeException.class)
GenericHandler<String> failedHandler() {
    return (payload, headers) -> {
        EaiAckNackOut eaiAckNackOut = new EaiAckNackOut();
        eaiAckNackOut.setStatus("foo");
        eaiAckNackOutRepository.save(eaiAckNackOut);
        log.info("FailedHandler processing: {}", payload);
        throw new RuntimeException("Error!"); // 触发当前事务回滚
    };
}

@Bean
@Transactional(rollbackFor = RuntimeException.class)
GenericHandler<String> successHandler() {
    return (payload, headers) -> {
        EaiAckNackOut eaiAckNackOut = new EaiAckNackOut();
        eaiAckNackOut.setStatus("bar");
        eaiAckNackOutRepository.save(eaiAckNackOut);
        log.info("SuccessHandler processing: {}", payload);
        return "Success";
    };
}

方式2：集成流中配置事务拦截器 #

通过Spring Integration的TransactionInterceptor为处理器添加事务支持，更贴合集成流配置风格：

@Bean
TransactionInterceptor transactionInterceptor(PlatformTransactionManager transactionManager) {
    TransactionInterceptor interceptor = new TransactionInterceptor();
    DefaultTransactionAttribute attribute = new DefaultTransactionAttribute();
    attribute.setRollbackRules(Collections.singletonList(new RollbackRuleAttribute(RuntimeException.class)));
    interceptor.setTransactionManager(transactionManager);
    interceptor.setTransactionAttributes(Collections.singletonMap("*", attribute));
    return interceptor;
}

// 在scatterFlow中为每个处理器绑定事务拦截器
@Bean
IntegrationFlow scatterFlow(TransactionInterceptor transactionInterceptor) {
    return f -> f.publishSubscribeChannel(scatterGatherChannel,
            s -> s.subscribe(sf -> sf.handle(failedHandler()).advice(transactionInterceptor))
                    .subscribe(sf -> sf.handle(successHandler()).advice(transactionInterceptor))
    );
}

配置后，failedHandler抛出异常时自身的数据库操作会回滚，但successHandler因异步独立执行，事务仍会正常提交。

二、实现全局事务一致性（全成功或全回滚） #

如果业务要求只要一个处理器失败，所有操作都回滚，异步并行的PublishSubscribeChannel无法满足（已提交的事务无法回滚），需调整方案：

方案1：使用同步PublishSubscribeChannel #

去掉TaskExecutor，让所有处理器同步执行，用全局事务包裹所有操作：

@Bean
public PublishSubscribeChannel scatterGatherChannel() {
    return MessageChannels.publishSubscribe().applySequence(true).get();
}

// 用事务拦截器包裹整个scatter-gather流程
@Bean
IntegrationFlow distributionFlow(PlatformTransactionManager transactionManager) {
    PollerSpec pollerSpec = Pollers.fixedDelay(Duration.ofSeconds(5))
            .maxMessagesPerPoll(1);

    return IntegrationFlows
            .from(msgSource(), c -> c.poller(pollerSpec).autoStartup(true))
            .log("org.eib.cms.connectivity.job.DistributionScatterGather", m -> "retrieved entity: " + m)
            .handle((payload, headers) -> payload, e -> e.advice(transactionInterceptor(transactionManager)))
            .scatterGather(
                    scatterGatherChannel,
                    gatherer -> gatherer.outputProcessor(new SimpleMessageGroupProcessor()),
                    scatterGather -> scatterGather.errorChannel("scatterGatherErrorChannel").gatherTimeout(10000)
            )
            .log("org.eib.cms.connectivity.job.DistributionScatterGather", m -> "Finished: " + m)
            .get();
}

此方式下所有处理器在同一事务中执行，任意处理器抛异常都会触发全局回滚，但会失去异步并行的性能优势。

方案2：引入分布式事务（强制异步场景） #

若必须保持异步并行且需要全局一致性，需引入Seata、Atomikos等分布式事务框架，让每个异步线程的事务参与全局事务，任意分支失败时触发全局回滚（会增加系统复杂度）。

• 确保项目已正确配置事务管理器（如JpaTransactionManager或DataSourceTransactionManager）
• 默认仅RuntimeException和Error会触发事务回滚，受检异常需通过rollbackFor指定
• 异步场景下，scatterGatherErrorChannel仅能捕获失败任务的异常，无法影响已完成任务的事务

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