如何基于Spring Cloud Stream Kafka实现Kafka @RetryableTopic？

spring-kafka提供的@RetryableTopic本质是通过自动创建分级重试主题、死信主题，配合主题级消息滞留延迟实现重试间隔控制，消费线程不会被长时间阻塞。spring-cloud-stream-binder-kafka默认内置的重试是消费线程内同步阻塞重试，如果未正确配置退避参数，就会出现重试逻辑正常执行但无间隔的问题，和@RetryableTopic的行为不一致。

两种方案都可以100%对齐@RetryableTopic的可靠重试+自定义延迟效果，按需选择即可。

方案1：直接复用@RetryableTopic注解（推荐，和原生spring-kafka体验完全一致） #

spring-cloud-stream-binder-kafka底层依赖spring-kafka实现消费逻辑，只要关闭binder自带的冲突重试配置，就可以直接在消费方法上使用@RetryableTopic，配置逻辑和原生用法无差异。

• 第一步：修改配置关闭binder层默认重试，避免双层重试冲突

spring:
  cloud:
    stream:
      kafka:
        bindings:
          <替换为实际消费binding名称>:
            consumer:
              enable-dlq: false # 关闭binder默认死信逻辑，交给@RetryableTopic托管
      bindings:
        <替换为实际消费binding名称>:
          consumer:
            max-attempts: 1 # binder层面不做重试，所有重试逻辑走@RetryableTopic

• 第二步：在消费方法上添加@RetryableTopic注解配置重试规则
  如果是新版函数式编程模型，直接把注解加在消费Bean方法上即可：

import org.springframework.kafka.annotation.RetryableTopic;
import org.springframework.kafka.annotation.Backoff;
import org.springframework.kafka.annotation.DltStrategy;
import org.springframework.kafka.annotation.TopicSuffixingStrategy;
import org.springframework.messaging.Message;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;
import java.util.function.Consumer;

@Component
public class BizConsumer {

    @RetryableTopic(
            attempts = "5", // 总消费次数（含首次正常消费）
            backoff = @Backoff(delay = 1000, multiplier = 2.0), // 退避规则：首次延迟1s，后续间隔翻倍
            dltStrategy = DltStrategy.FAIL_ON_ERROR,
            topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE
    )
    @Bean
    public Consumer<Message<BizPayload>> consumeBizMsg() {
        return message -> {
            // 业务消费逻辑，抛出异常即触发重试
            System.out.printf("消费消息：%s%n", message.getPayload());
        };
    }
}

如果是老版本用@StreamListener定义消费逻辑，直接把@RetryableTopic加在对应监听方法上即可，其他配置不变。

注意：Spring Cloud 2021.0.x及以上版本自带的spring-kafka版本完全兼容该用法，不需要额外引入依赖。

方案2：Binder原生配置实现（无注解侵入） #

如果不想使用spring-kafka的专属注解，可通过binder自带的非阻塞重试配置实现和@RetryableTopic完全等价的能力，binder会自动创建分级重试主题、死信主题并维护延迟路由逻辑：

spring:
  cloud:
    stream:
      kafka:
        bindings:
          <替换为实际消费binding名称>:
            consumer:
              non-blocking-retry: true # 开启主题级非阻塞重试，和@RetryableTopic原理一致
              enable-dlq: true
              max-attempts: 5 # 总消费次数
              # 退避延迟配置，对应1s、2s、4s、8s的重试间隔
              back-off-initial-interval: 1000
              back-off-multiplier: 2.0
              back-off-max-interval: 8000
              # 主题后缀规则和@RetryableTopic默认对齐
              retry-topic-suffix: "-retry"
              dlq-topic-suffix: "-dlt"
              delay-expression: "headers['kafka_retry_backoff']"
      bindings:
        <替换为实际消费binding名称>:
          consumer:
            max-attempts: 5 # 和kafka consumer层的重试次数保持一致

如果之前配置重试后没有达到预期延迟，基本是以下原因：

• 未开启non-blocking-retry: true，走默认的容器内阻塞重试但未配置backoff参数，导致消息无间隔立刻重试
• 同时开启了binder自带重试和@RetryableTopic，两层重试逻辑冲突，延迟参数被覆盖
• 重试主题的消息滞留时间配置错误，消息进入重试主题后被立刻投回主消费主题

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