Spring Cloud Streams(Elmhurst/2.0)异常捕获与处理方案咨询

最近在基于Spring Cloud Streams Elmhurst.RELEASE版本做消息处理时，踩了个异常处理的大坑——主处理方法抛出异常后，完全没法按文档预期在错误通道捕获到异常，折腾了好久才找到合适的解决方案，在这里分享给大家：

问题场景 #

当我的消息处理方法抛出异常时，无论是全局的errorChannel还是特定的input.myGroup.errors通道，都接收不到ErrorMessage。我先后尝试了：

• 用Spring Integration的@ServiceActivator监听errorChannel
• 用Spring Cloud Streams的@StreamListener监听errorChannel
• 调整errorChannelEnabled、max-retries等配置项
• 配置bindings.error.destination指定错误消息目标

所有尝试都没效果，唯一能生效的是在Kafka binder配置里开启enableDlq，但默认DLQ里的消息只有带完整堆栈的头部信息，我总不能去解析堆栈来获取异常类型和消息吧？这完全不符合我的需求——我需要把包含实际异常类型、异常消息以及原消息内容的结构化数据发送到DLQ。

走过的弯路 #

• 一开始想过用手动try/catch包裹处理逻辑，但这直接抛弃了SCS框架的异常处理能力，完全违背了框架设计的初衷，肯定不能这么干。
• 后来找到了Gary Russell的自定义DLQ处理示例，但在SCS Elmhurst/2.0 + Spring Kafka binder的环境下根本跑不起来，我还专门搭了个仓库复现问题。

最终解决方案 #

在Gary的帮助下先解决了配置环境的问题，但默认DLQ的消息格式还是不符合要求，最后我采用了监听全局errorChannel，自定义错误消息并发送到指定绑定通道的方案，完美解决了问题：

核心实现代码 #

首先是主应用的消息处理逻辑：

@SpringBootApplication
@EnableBinding(Processor.class)
public class DemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(DemoApplication.class, args);
    }

    @Autowired
    private MessagingTemplate messagingTemplate;

    @StreamListener(Processor.INPUT)
    @SendTo(Processor.OUTPUT)
    public String process(String message) {
        externalService();
        return message.toUpperCase();
    }

    private void externalService() {
        throw new RuntimeException("An external call failed");
    }
}

然后是自定义的错误处理逻辑，用@ServiceActivator监听errorChannel，封装自定义消息后发送到DLQ通道：

@ServiceActivator(inputChannel = "errorChannel")
public void handleError(ErrorMessage errorMessage) {
    // 提取原异常和原消息
    Throwable exception = errorMessage.getPayload();
    Object originalMessage = errorMessage.getHeaders().getOrDefault(MessageHeaders.PAYLOAD, "Unknown message");
    
    // 构建结构化的错误消息
    Map<String, Object> dlqMessage = new HashMap<>();
    dlqMessage.put("originalMessage", originalMessage);
    dlqMessage.put("exceptionType", exception.getClass().getCanonicalName());
    dlqMessage.put("exceptionMessage", exception.getMessage());
    dlqMessage.put("timestamp", System.currentTimeMillis());

    // 发送到自定义DLQ绑定通道
    messagingTemplate.send(Processor.DLQ_OUTPUT, 
        MessageBuilder.withPayload(dlqMessage)
            .copyHeaders(errorMessage.getHeaders())
            .build());
}

配套配置（application.yml） #

需要确保开启错误通道，同时关闭默认的DLQ：

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: input-topic
          group: myGroup
          consumer:
            error-channel-enabled: true # 必须开启，否则errorChannel不会收到消息
            max-retries: 0 # 可以根据需求设置重试次数，这里直接进入错误通道
        output:
          destination: output-topic
        dlq-output:
          destination: dlq-topic # 自定义的DLQ主题
      kafka:
        binder:
          enableDlq: false # 关闭默认DLQ，使用自定义逻辑

这样一来，当消息处理抛出异常时，我们就能在自定义DLQ里拿到结构化的错误信息，不用再去解析堆栈了，完全满足需求。

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