Spring Integration：如何阻止错误传播并避免轮询器事务回滚？

我有一个基于Spring Integration的服务，从数据库获取JSON数组（每个数组代表一笔交易），这些数组会转换为目标数据库的操作，因此在@ServiceActivator上配置了adviceChain以保证事务。

• 当数据库操作失败时，数据需进入“数据修复中心（data hospital）”进行人工修正，此时需避免轮询器事务回滚，防止数据被重复拉取。
• 同时需要区分异常类型，例如目标库连接错误这类异常需触发轮询器事务回滚。

原始实现代码 #

@Bean
public IntegrationFlow pollingFlow(@Qualifier(PgSqlDataSourceConfig.PSQL_TRAN_MANAGER) final TransactionManager pgsqlTM) {
    return IntegrationFlows.from(jdbcMessageSource(),
                    c -> c.poller(Pollers.fixedDelay(properties.getRateMs()).transactional(pgsqlTM)))
            .transform(controleGetJsDados())
            .channel(SybaseJsonSinkService.SYBASE_SINK)
            .get();
}

@Bean
public IntegrationFlow sinkFlow() {
    return IntegrationFlows.from(SYBASE_SINK)
            .transform(jsonToDbOpTransformer())
            .channel(SYBASE_SINK_SQL)
            .get();
}

@Bean
public PublishSubscribeChannel errorChannel() {
    return new PublishSubscribeChannel();
}

@Bean
public IntegrationFlow errorFlow() {
    return IntegrationFlows.from(SYBASE_ERROR)
            .routeByException(r -> r.channelMapping(HospitalException.class, SYBASE_HOSPITAL)
                                    .channelMapping(Roman8Exception.class, SYBASE_SINK_ROMAN8)
                                    .defaultOutputToParentFlow())
            .get();
}

@ServiceActivator(inputChannel = SYBASE_SINK_SQL, adviceChain = {TRAN_INTERCEPTOR})
public void sinkSql(final Message<DbTransaction> message) {
    try {
        // 目标数据库操作逻辑
    } catch(Exception ex) {
        throw new HospitalException(message.getPayload(), ex);
    }
}

@ServiceActivator(inputChannel = SYBASE_HOSPITAL)
public void sybaseHospital(final ErrorMessage errorMessage) {
    // 将数据存入数据修复中心
}

尝试过的无效配置 #

public TransactionInterceptor transactionInterceptor(final TransactionManager tm) {
    RuleBasedTransactionAttribute rbta = new RuleBasedTransactionAttribute();
    List<RollbackRuleAttribute> rollbackRules = new ArrayList<>();
    rollbackRules.add(new NoRollbackRuleAttribute(Roman8Exception.class));
    rollbackRules.add(new NoRollbackRuleAttribute(HospitalException.class));
    rbta.setRollbackRules(rollbackRules);

    return new TransactionInterceptorBuilder(true)
            .transactionManager(tm)
            .isolation(Isolation.READ_COMMITTED)
            .propagation(Propagation.REQUIRES_NEW)
            .transactionAttribute(rbta)
            .build();
}


@Bean
public IntegrationFlow pollingFlow(@Qualifier(PgSqlDataSourceConfig.PSQL_TRAN_MANAGER) final TransactionManager pgsqlTM) {
    return IntegrationFlows.from(jdbcMessageSource(),
                    c -> c.poller(Pollers.fixedDelay(properties.getRateMs())
                                         .transactional(transactionInterceptor(pgsqlTM)).taskExecutor(executor())))
            .transform(controleGetJsDados())
            .channel(SybaseJsonSinkService.SYBASE_SINK)
            .get();
}

核心问题在于轮询器的事务和下游目标库操作的事务需要解耦，同时精准控制不同异常对轮询事务的影响：

1. 拆分事务边界 #

轮询器的事务仅管控源数据库的读取操作，下游目标库的操作使用独立事务。这样目标库操作的异常不会直接影响轮询事务，除非是需要回滚的严重异常（如目标库连接失败）。

2. 调整事务拦截器的作用范围 #

不要给轮询器绑定包含NoRollbackRule的拦截器，而是将该拦截器仅应用在目标库操作的@ServiceActivator上。轮询器保持基础的事务配置，仅在捕获到指定严重异常时才回滚。

3. 通过异常路由控制轮询事务回滚 #

• 对于HospitalException和Roman8Exception，在错误流中处理后，不要将异常抛回轮询器的事务上下文。
• 对于目标库连接错误这类严重异常，需要将异常重新抛出，触发轮询器的事务回滚。

修改后的关键配置 #

轮询器恢复基础事务配置

@Bean
public IntegrationFlow pollingFlow(@Qualifier(PgSqlDataSourceConfig.PSQL_TRAN_MANAGER) final TransactionManager pgsqlTM) {
    return IntegrationFlows.from(jdbcMessageSource(),
                    c -> c.poller(Pollers.fixedDelay(properties.getRateMs())
                                         .transactional(pgsqlTM)
                                         .taskExecutor(executor()))) // 保持异步执行
            .transform(controleGetJsDados())
            .channel(SybaseJsonSinkService.SYBASE_SINK)
            .get();
}

调整事务拦截器，仅作用于目标库操作

@Bean
public TransactionInterceptor targetDbTransactionInterceptor(final TransactionManager targetTm) { // 使用目标库的事务管理器
    RuleBasedTransactionAttribute rbta = new RuleBasedTransactionAttribute();
    List<RollbackRuleAttribute> rollbackRules = new ArrayList<>();
    // 目标库操作自身的事务规则：仅在特定异常时回滚
    rollbackRules.add(new RollbackRuleAttribute(ConnectionException.class)); // 目标库连接异常回滚自身事务
    rollbackRules.add(new NoRollbackRuleAttribute(HospitalException.class));
    rollbackRules.add(new NoRollbackRuleAttribute(Roman8Exception.class));
    rbta.setRollbackRules(rollbackRules);

    return new TransactionInterceptorBuilder(true)
            .transactionManager(targetTm)
            .isolation(Isolation.READ_COMMITTED)
            .propagation(Propagation.REQUIRES_NEW) // 开启独立事务
            .transactionAttribute(rbta)
            .build();
}

// 修改ServiceActivator，使用目标库的事务拦截器
@ServiceActivator(inputChannel = SYBASE_SINK_SQL, adviceChain = {TARGET_DB_TRAN_INTERCEPTOR})
public void sinkSql(final Message<DbTransaction> message) {
    try {
        // 目标数据库操作逻辑
    } catch(ConnectionException ex) {
        // 连接异常直接抛出，触发轮询事务回滚
        throw ex;
    } catch(Exception ex) {
        throw new HospitalException(message.getPayload(), ex);
    }
}

调整错误流，避免将无需回滚的异常抛回轮询器

@Bean
public IntegrationFlow errorFlow() {
    return IntegrationFlows.from(SYBASE_ERROR)
            .routeByException(r -> r.channelMapping(HospitalException.class, SYBASE_HOSPITAL)
                                    .channelMapping(Roman8Exception.class, SYBASE_SINK_ROMAN8)
                                    .defaultOutputChannel("errorChannel")) // 严重异常走默认错误通道，触发轮询回滚
            .get();
}

关键说明 #

• 轮询器的事务仅负责源数据的读取，一旦数据被成功读取并发送到下游，除非捕获到需要回滚的异常（如目标库连接失败），否则不会回滚，避免重复拉取。
• 目标库操作使用独立事务，通过事务拦截器控制自身的回滚规则，同时通过异常类型区分是否需要触发轮询事务的回滚。
• 错误流中处理无需回滚的异常后，终止异常传播；严重异常则继续传播到轮询器的事务上下文，触发回滚。

内容的提问来源于stack exchange，提问作者Thiago Sayão