如何在Spring Cloud AWS Kinesis Stream中实现通用异常处理器

当然可以实现这种通用的异常处理机制！Spring Cloud Stream本身就提供了灵活的错误处理扩展点，结合Binder配置、自定义Transformer和Router，完全能替代你现在手动捕获错误的方式，而且能覆盖所有通道。下面我一步步给你拆解实现方案：

首先要让所有输入通道的异常消息能统一流转到错误处理链路。通过开启error-channel-enabled配置，Spring Cloud Stream会将消费者处理失败的消息自动发送到全局errorChannel（也可以自定义通道名）。以Kinesis Binder为例，在application.yml中配置：

spring:
  cloud:
    stream:
      # 全局默认配置，避免重复配置每个通道
      default:
        consumer:
          error-channel-enabled: true # 开启错误通道转发
          enable-dlq: false # 禁用默认DLQ，改用自定义处理
          retry:
            enabled: true # 可选：先重试再进入错误通道
            max-attempts: 3
            back-off:
              initial-interval: 1000
      bindings:
        # 你的业务输入通道示例
        input1-in-0:
          destination: your-kinesis-stream-1
          binder: kinesis
        input2-in-0:
          destination: your-kinesis-stream-2
          binder: kinesis
      kinesis:
        binder:
          auto-create-stream: true

Spring默认的ErrorMessage包含原始异常和消息，但结构比较底层，我们可以用Transformer把它转换成统一的自定义格式，方便后续路由处理：

import org.springframework.cloud.stream.annotation.Transformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.MessageBuilder;

public class ErrorMessageTransformer {

    @Transformer(inputChannel = "errorChannel", outputChannel = "transformedErrorChannel")
    public Message<?> transformErrorMessage(ErrorMessage errorMessage) {
        // 提取核心错误信息
        Throwable exception = errorMessage.getPayload();
        Message<?> originalMsg = errorMessage.getOriginalMessage();
        
        // 构造标准化错误负载
        CustomErrorPayload payload = CustomErrorPayload.builder()
                .exceptionType(exception.getClass().getSimpleName())
                .errorMsg(exception.getMessage())
                .originalPayload(originalMsg != null ? originalMsg.getPayload() : null)
                .sourceChannel(originalMsg != null ? 
                    originalMsg.getHeaders().get("spring.cloud.stream.binding.name").toString() : 
                    "unknown-channel")
                .build();
        
        // 返回带标准负载的消息，保留原头部信息
        return MessageBuilder.withPayload(payload)
                .copyHeaders(errorMessage.getHeaders())
                .build();
    }
}

// 自定义错误负载类（用Lombok简化代码）
@Data
@Builder
public class CustomErrorPayload {
    private String exceptionType;
    private String errorMsg;
    private Object originalPayload;
    private String sourceChannel;
}

接下来用Router根据错误的来源通道或异常类型，将消息分发到不同的业务错误处理通道：

import org.springframework.cloud.stream.annotation.Router;
import org.springframework.messaging.Message;

public class ErrorMessageRouter {

    @Router(inputChannel = "transformedErrorChannel")
    public String routeErrorMessage(Message<CustomErrorPayload> errorMsg) {
        CustomErrorPayload payload = errorMsg.getPayload();
        
        // 按来源通道路由
        switch (payload.getSourceChannel()) {
            case "input1-in-0":
                return "error-handler-input1";
            case "input2-in-0":
                return "error-handler-input2";
            default:
                return "error-handler-default";
        }
        
        // 也可以按异常类型路由，比如：
        // if ("IllegalArgumentException".equals(payload.getExceptionType())) {
        //     return "validation-error-channel";
        // }
    }
}

然后在配置类中注册Transformer和Router：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ErrorHandlingConfig {

    @Bean
    public ErrorMessageTransformer errorMessageTransformer() {
        return new ErrorMessageTransformer();
    }

    @Bean
    public ErrorMessageRouter errorMessageRouter() {
        return new ErrorMessageRouter();
    }
}

最后为每个路由目标通道配置对应的消费者，处理不同场景的错误：

spring:
  cloud:
    stream:
      bindings:
        error-handler-input1-in-0:
          destination: kinesis-error-stream-input1
          binder: kinesis
          consumer:
            group: error-handler-group-input1
        error-handler-input2-in-0:
          destination: kinesis-error-stream-input2
          binder: kinesis
          consumer:
            group: error-handler-group-input2
        error-handler-default-in-0:
          destination: kinesis-error-stream-default
          binder: kinesis
          consumer:
            group: error-handler-group-default

编写对应的错误处理逻辑：

import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.Payload;

public class ErrorHandlerConsumers {

    @StreamListener("error-handler-input1")
    public void handleInput1Errors(@Payload CustomErrorPayload payload) {
        // 针对input1的错误处理：日志记录、告警、数据修复等
        System.err.printf("[INPUT1 ERROR] %s: %s%n", payload.getExceptionType(), payload.getErrorMsg());
    }

    @StreamListener("error-handler-input2")
    public void handleInput2Errors(@Payload CustomErrorPayload payload) {
        // 针对input2的错误处理
        System.err.printf("[INPUT2 ERROR] %s: %s%n", payload.getExceptionType(), payload.getErrorMsg());
    }

    @StreamListener("error-handler-default")
    public void handleDefaultErrors(@Payload CustomErrorPayload payload) {
        // 未匹配到特定通道的默认错误处理
        System.err.printf("[DEFAULT ERROR] From %s: %s%n", payload.getSourceChannel(), payload.getErrorMsg());
    }
}

• 如果需要对特定异常进行重试或忽略，可以在Transformer中提前过滤，或者在Router中调整路由规则。
• 若要监控错误链路，可以在Transformer或Router中添加埋点日志，方便排查问题。

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