Spring Integration子流中JDBCMessageHandler异常处理策略如何配置

实现方案 #

你注释掉的publishSubscribeChannel.errorHandler仅用于自定义订阅者异常的封装逻辑，默认会把异常重新向上抛出，无法实现拦截终止异常的需求。可以通过以下两种方式实现子流层级的异常捕获，异常不会向上传播到poller，同时保留完整的消息上下文：

方案1：为JDBC处理器配置异常处理通知 #

给每个JDBCMessageHandler添加ExpressionEvaluatingRequestHandlerAdvice，异常发生时直接路由到专用错误处理流，处理完成后终止流程：

// 定义通用异常处理通知
@Bean
public Advice jdbcHandlerErrorAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    // 异常发生时路由到你的错误处理流入口通道
    advice.setFailureChannelName("customErrorFlowInputChannel");
    // 捕获异常不向上抛出
    advice.setTrapException(true);
    return advice;
}

修改子流代码，给每个JDBC处理器绑定该通知：

return flowDef
    .filter(getFilterExpression(rule)).channel(new DirectChannel())
    .handle(inboundAdapter)
    .split(insertDeleteSplitter)
    .publishSubscribeChannel(c -> 
            c.subscribe(s -> s
                .filter ("....")
                .transform(genericTransformer)
                .handle(insertUpdateMessageHandler(rule), e -> e.advice(jdbcHandlerErrorAdvice())))
            .subscribe(s -> s
                .filter("....")
                .transform(genericTransformer)
                .handle(deleteMessageHandler(rule), e -> e.advice(jdbcHandlerErrorAdvice())))
            .subscribe(sub -> sub
                .handle(cleanupMessageHandler(rule), e -> e.advice(jdbcHandlerErrorAdvice())))
    );

你可以单独定义错误处理流，从接收到的ErrorMessage中获取原始消息、异常详情，完成故障记录、核心数据落地、清理流程等操作。

方案2：子流外层嵌套网关拦截所有异常 #

将当前业务子流封装到网关内部，在网关层配置专属错误通道，所有子流抛出的异常都会直接进入该错误通道处理：

return flowDef
    .filter(getFilterExpression(rule)).channel(new DirectChannel())
    .handle(inboundAdapter)
    .split(insertDeleteSplitter)
    .gateway(gw -> gw
        .publishSubscribeChannel(c -> 
                c.subscribe(s -> s
                    .filter ("....")
                    .transform(genericTransformer)
                    .handle(insertUpdateMessageHandler(rule)))
                .subscribe(s -> s
                    .filter("....")
                    .transform(genericTransformer)
                    .handle(deleteMessageHandler(rule)))
                .subscribe(sub -> sub
                    .handle(cleanupMessageHandler(rule)))
        )
        // 配置网关专属错误通道
        .errorChannel("customErrorFlowInputChannel")
    );

两种方案均满足需求，不需要自定义包装JDBC处理器，也不会影响原有事务逻辑。

内容的提问来源于stack exchange，提问作者pojo-guy