Spring Cloud Stream中排除CustomException其余异常转入DLQ的实现

要实现仅将CustomException以外的异常消息转发到DLQ，可通过自定义错误处理器结合Spring Cloud Stream的SQS绑定器能力完成，以下是具体步骤和代码示例：

1. 自定义错误处理器 #

实现ErrorHandler接口，在处理器中判断异常类型，仅将非CustomException的异常消息转发到DLQ：

import org.springframework.cloud.stream.binder.SqsMessageHeaders;
import org.springframework.cloud.stream.messaging.ErrorChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.util.ErrorHandler;

@Component("customSqsErrorHandler")
public class CustomSqsErrorHandler implements ErrorHandler {

    private final ErrorChannel errorChannel;

    public CustomSqsErrorHandler(ErrorChannel errorChannel) {
        this.errorChannel = errorChannel;
    }

    @Override
    public void handleError(Throwable throwable) {
        if (throwable instanceof org.springframework.messaging.MessagingException messagingException) {
            Message<?> originalMsg = messagingException.getFailedMessage();
            if (originalMsg != null) {
                Throwable rootCause = getRootCause(throwable);
                // 仅转发非CustomException的异常消息到DLQ
                if (!(rootCause instanceof CustomException)) {
                    Message<?> dlqMsg = MessageBuilder.fromMessage(originalMsg)
                            .setHeader(SqsMessageHeaders.SQS_DESTINATION_SUFFIX, "-dlq")
                            .build();
                    errorChannel.send(dlqMsg);
                }
                // 对于CustomException，可在此添加日志记录或其他业务处理
            }
        }
    }

    // 提取异常根因，避免包装类干扰判断
    private Throwable getRootCause(Throwable throwable) {
        Throwable cause;
        while ((cause = throwable.getCause()) != null) {
            throwable = cause;
        }
        return throwable;
    }
}

2. 配置消费者绑定 #

在application.yml中指定自定义错误处理器，并根据需求配置重试策略：

spring:
  cloud:
    stream:
      bindings:
        input: # 你的消费者绑定名称
          destination: your-main-queue # 主队列名称
          group: your-consumer-group
          binder: aws-sqs
      binders:
        aws-sqs:
          type: sqs
          environment:
            spring:
              cloud:
                aws:
                  sqs:
                    region: your-aws-region # 例如 us-east-1
      sqs:
        bindings:
          input:
            consumer:
              error-handler-bean-name: customSqsErrorHandler # 指定自定义错误处理器
              retry:
                enabled: true # 开启重试（可选，根据业务需求调整）
                max-attempts: 3
                back-off:
                  initial-interval: 1000ms

关键说明 #

• DLQ队列关联：确保主队列对应的DLQ队列（名称格式为your-main-queue-dlq）已在AWS控制台创建，或配置Spring Cloud Stream自动创建队列（需赋予SQS队列创建权限）。
• 异常根因判断：通过getRootCause方法提取真实异常，避免被MessagingException等包装类干扰类型判断。
• CustomException处理：对于CustomException，可在处理器中添加日志、告警或其他业务逻辑，不触发DLQ转发。

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