Spring Integration：ServiceActivator异常统一处理方案咨询

1. 用ExpressionEvaluatingRequestHandlerAdvice统一捕获@ServiceActivator异常 #

这是Spring Integration官方推荐的端点异常处理方式，能精准控制每个消息处理端点的异常逻辑，无需手动写try/catch。

步骤1：定义全局异常处理Advice #

@Bean
public ExpressionEvaluatingRequestHandlerAdvice errorHandlingAdvice(MessageChannel errorMessageChannel) {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    // 开启异常捕获，避免异常向上传播中断流程
    advice.setTrapException(true);
    // 异常触发时，设置ERROR头为true，并携带异常信息
    advice.setOnFailureExpressionString(
            "headers.put('ERROR', true); " +
            "headers.put('EXCEPTION', rootCause); " +
            "payload"
    );
    // 处理完异常后，直接把消息转发到你的错误通道
    advice.setFailureChannel(errorMessageChannel);
    return advice;
}

步骤2：给@ServiceActivator绑定该Advice #

给需要异常处理的服务激活器添加adviceChain属性，复用上面的全局Advice：

@ServiceActivator(inputChannel = "newRequestChannel", adviceChain = "errorHandlingAdvice")
public void handleNewRequest(Message<?> message) {
    // 业务逻辑，抛出的异常会被自动捕获处理
}

@ServiceActivator(inputChannel = "inProgressChannel", adviceChain = "errorHandlingAdvice")
public void handleInProgress(Message<?> message) {
    // 业务逻辑
}

当@ServiceActivator抛出异常时，Advice会自动标记消息为错误状态，并转发到errorMessageChannel，完全符合你现有错误监听流程的逻辑。

2. 全局通道拦截器批量处理指定通道异常 #

如果你想一次性覆盖多个通道的异常处理，不用逐个修改@ServiceActivator，可以用全局通道拦截器：

@Component
@GlobalChannelInterceptor(patterns = {"newRequestChannel", "inProgressChannel"})
public class ChannelErrorInterceptor implements ChannelInterceptor {

    private final MessageChannel errorMessageChannel;

    public ChannelErrorInterceptor(MessageChannel errorMessageChannel) {
        this.errorMessageChannel = errorMessageChannel;
    }

    @Override
    public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
        if (ex != null) {
            // 构造带错误标记的消息，发送到错误通道
            Message<?> errorMsg = MessageBuilder.fromMessage(message)
                    .setHeader("ERROR", true)
                    .setHeader("EXCEPTION", ex.getCause())
                    .build();
            errorMessageChannel.send(errorMsg);
        }
    }
}

拦截器会监听指定通道的消息处理完成事件，一旦出现异常就自动转发错误消息，适合批量统一处理的场景。

3. 异步通道用ErrorHandlingTaskExecutor处理异常 #

如果你的newRequestChannel和inProgressChannel是异步类型（比如ExecutorChannel），可以给线程池绑定错误处理器：

步骤1：定义带错误处理的线程池 #

@Bean
public TaskExecutor errorHandlingTaskExecutor(MessageChannel errorMessageChannel) {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(10);
    executor.setQueueCapacity(25);
    // 自定义错误处理器，捕获线程池中的异常
    executor.setErrorHandler(throwable -> {
        if (throwable instanceof MessagingException) {
            MessagingException msgEx = (MessagingException) throwable;
            Message<?> errorMsg = MessageBuilder.fromMessage(msgEx.getFailedMessage())
                    .setHeader("ERROR", true)
                    .setHeader("EXCEPTION", msgEx.getCause())
                    .build();
            errorMessageChannel.send(errorMsg);
        }
    });
    executor.initialize();
    return executor;
}

步骤2：把通道配置为异步通道 #

@Bean
public MessageChannel newRequestChannel(TaskExecutor errorHandlingTaskExecutor) {
    return new ExecutorChannel(errorHandlingTaskExecutor);
}

@Bean
public MessageChannel inProgressChannel(TaskExecutor errorHandlingTaskExecutor) {
    return new ExecutorChannel(errorHandlingTaskExecutor);
}

异步场景下的异常会被线程池的错误处理器捕获，自动转发到错误通道。

4. 主流程添加全局错误分支 #

如果希望整个IntegrationFlow的所有未处理异常都统一走错误通道，可以在主流程末尾添加全局错误配置：

public IntegrationFlow flow(MessageChannel errorMessageChannel) {
    return IntegrationFlows.from(commonMonitoringMessageChannel())
            .route(Message.class,
                    message -> message.getHeaders().get("ERROR"),
                    mapping -> mapping.channelMapping(true, errorMessageChannel)
                            .subFlowMapping(false, sf -> sf.route(
                                    Message.class,
                                    message -> message.getHeaders().get("STAGE"),
                                    subMapping -> subMapping
                                            .channelMapping("NEW_REQUEST", newRequestChannel())
                                            .channelMapping("IN_PROGRESS", inProgressChannel())
                            ))
            )
            // 捕获整个流程中未被处理的异常，直接转发到错误通道
            .errorChannel(errorMessageChannel)
            .get();
}

注意这种方式需要去掉@ServiceActivator中的手动try/catch，让异常能向上传播到主流程的错误分支。

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