Spring Cloud Stream Kafka：阻止特定异常重试及进入DLQ的方案

一、直接捕获异常忽略是否可行？ #

可行，但存在明显的优缺点：

• 优点：实现简单、快速生效。在业务代码中捕获DuplicateReverseException且不抛出，结合你的配置ackEachRecord: true，框架会判定消息处理成功并自动提交offset，既不会触发重试，也不会将消息发送到DLQ。
• 缺点：业务逻辑与错误处理逻辑强耦合，后续若需调整异常类型或处理策略，必须修改业务代码，灵活性较差。

修改后的消费者代码示例：

@Bean
public Consumer<Message<News>> news() {
    return message -> {
        try {
            if (message.getPayload().getSource().equals("1")) {
                log.error("1");
                throw new DuplicateReverseException();
            } else {
                log.error("2");
                throw new ApplicationException();
            }
        } catch (DuplicateReverseException e) {
            log.warn("忽略DuplicateReverseException，消息不再重试或进入DLQ");
            // 无需额外操作，框架自动提交offset
        }
    };
}

二、Spring提供的更优解决方案 #

推荐使用Spring Cloud Stream的错误扩展机制，将错误处理与业务逻辑解耦，以下两种方案更优雅：

方案1：自定义ConsumerErrorHandler #

创建全局错误处理器，针对特定异常类型跳过DLQ发送：

1. 定义自定义错误处理器Bean：

@Slf4j
@Bean("customConsumerErrorHandler")
public ConsumerErrorHandler customConsumerErrorHandler(DeadLetterPublishingRecoverer defaultDlqRecoverer) {
    return (message, throwable) -> {
        Throwable rootCause = ExceptionUtils.getRootCause(throwable);
        if (rootCause instanceof DuplicateReverseException) {
            log.warn("检测到DuplicateReverseException，跳过DLQ发送");
            // 手动提交offset（确保消息不会重复消费）
            Acknowledgment acknowledgment = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
            if (acknowledgment != null) {
                acknowledgment.acknowledge();
            }
        } else {
            // 其他异常交给默认DLQ处理器
            defaultDlqRecoverer.accept(message, throwable);
        }
    };
}

2. 在配置中指定该错误处理器：

spring:
  cloud:
    stream:
      bindings:
        news-in-0:
          consumer:
            error-handler-name: customConsumerErrorHandler

方案2：自定义DeadLetterPublishingRecoverer #

直接修改DLQ发送逻辑，过滤不需要进入DLQ的异常：

1. 定义自定义DLQ恢复器Bean：

@Slf4j
@Bean("customDlqRecoverer")
public DeadLetterPublishingRecoverer customDlqRecoverer(KafkaTemplate<?, ?> kafkaTemplate) {
    DeadLetterPublishingRecoverer defaultRecoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
    return (consumerRecord, exception) -> {
        Throwable rootCause = ExceptionUtils.getRootCause(exception);
        if (!(rootCause instanceof DuplicateReverseException)) {
            // 非目标异常，正常发送到DLQ
            defaultRecoverer.accept(consumerRecord, exception);
        } else {
            log.warn("检测到DuplicateReverseException，跳过DLQ发送");
            // 手动提交offset
            if (consumerRecord instanceof Acknowledgment) {
                ((Acknowledgment) consumerRecord).acknowledge();
            }
        }
    };
}

2. 在配置中指定该恢复器：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          news-in-0:
            consumer:
              dlq-recoverer-bean-name: customDlqRecoverer

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