如何在Spring Cloud（Spring Integration）中正确配置全局errorChannel

问题背景 #

基于Kinesis Stream、spring-cloud-stream、spring-cloud-stream-binder-kinesis和spring-cloud-starter-sleuth搭建Web应用，需实现三级错误处理逻辑：

1. 正常消费处理消息流
2. 为每个流配置专属错误处理器，处理消息消费时的异常
3. 全局errorChannel捕获专属错误处理器抛出的所有异常

目前步骤1、2可正常运行，但专属错误处理器抛出test2异常后，全局errorChannel未输出预期日志，Kinesis适配器提示需使用errorChannel，但异常直接抛出未被全局通道处理。

当前配置属性：

spring:
  cloud:
    stream:
      function:
        definition: destination1
      bindings:
        destination1-in-0:
          destination: destination1
          group: group1

当前核心代码：

@Bean
public Consumer<MyStream> destination1() {
    return message -> {
        if (true) throw new RuntimeException("test");
        LOGGER.info("message: " + message);
    };
}

@ServiceActivator(inputChannel = "destination1.group1.errors")
public void errorHandle(Message<?> message) {
    LOGGER.error("Handling ERROR: " + message);
    throw new RuntimeException("test2");
}

@Bean
public PublishSubscribeChannel errorChannel() {
    ErrorHandler errorHandler = throwable -> LOGGER.error("Handling ERROR in errorChannel error handler: ");
    Executor executor = new ConcurrentTaskExecutor();
    PublishSubscribeChannel publishSubscribeChannel = new PublishSubscribeChannel(executor);
    publishSubscribeChannel.setErrorHandler(errorHandler);
    return publishSubscribeChannel;
}

原因分析 #

1. 自定义errorChannel Bean未被正确识别：Spring Cloud Stream默认使用名为errorChannel的全局错误通道，但直接创建PublishSubscribeChannel并设置ErrorHandler的方式不符合Spring Integration错误处理流程——通道的ErrorHandler仅处理通道自身发送/接收时的异常，而非接收全局错误消息。
2. 专属错误处理器未转发异常到全局通道：当前专属处理器直接抛出异常，该异常会被Kinesis binder的错误逻辑重新抛出，而非路由到errorChannel。

解决方案 #

1. 移除自定义的errorChannel Bean #

Spring会自动创建默认的errorChannel（类型为PublishSubscribeChannel），无需手动定义。

2. 修改专属错误处理器，将异常转发到全局errorChannel #

通过注入全局errorChannel，将捕获的异常封装为消息发送到全局通道，而非直接抛出：

@Autowired
@Qualifier("errorChannel")
private MessageChannel errorChannel;

@ServiceActivator(inputChannel = "destination1.group1.errors")
public void errorHandle(Message<?> message) {
    LOGGER.error("Handling ERROR: " + message);
    // 从错误消息中提取原始异常，转发到全局errorChannel
    if (message.getPayload() instanceof MessagingException messagingException) {
        errorChannel.send(MessageBuilder.withPayload(messagingException.getCause())
                .copyHeaders(message.getHeaders())
                .build());
    } else {
        errorChannel.send(message);
    }
}

3. 添加全局errorChannel的消息处理器 #

通过@ServiceActivator监听errorChannel，处理全局错误：

@ServiceActivator(inputChannel = "errorChannel")
public void globalErrorHandle(Throwable throwable) {
    LOGGER.error("Handling ERROR in global errorChannel: ", throwable);
}

验证效果 #

修改后，当消息处理抛出test异常时：

1. 专属错误处理器errorHandle先打印日志，并将异常转发到errorChannel
2. 全局处理器globalErrorHandle接收异常，打印预期的全局错误日志
3. 不再直接抛出test2异常，异常被全局通道正确处理

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