最近業務上用到了Spring Kafka，所以系統性的探索了下Spring Kafka的各種用法，發現了很多實用的特性，下面介紹下Spring Kafka的消息重試機制。

0. 前言

原生 Kafka 是不支持消息重試的。但是 Spring Kafka 2.7+ 封裝了 Retry Topic 這個功能。

1. @RetryableTopic

使用注解的方式啟用 Retry Topic，在 @KafkaListener 方法上添加 @RetryableTopic 即可：

@Slf4j
@Component
public class DemoConsumer {
    @RetryableTopic
    @KafkaListener(topics = "topic1", groupId = "group1")
    public void onMsg(ConsumerRecord< String, String > record) {
        log.info("topic: {}", record.topic());
        throw new RuntimeException("kafka exception");
    }

}

這樣就開啟了 Spring Kafka 的消息重試機制：默認重試 3 次，間隔為 1 秒。

我們在方法里模擬了拋出異常，運行后可以發現打印了 3 條日志，間隔時間大約為 1 秒，重試的topic為原topic加上后綴“-retry”

2022-11-12 12:14:10.230  INFO 1023 --- [ner#3-dlt-0-C-1] c.b.b.demo.retrytopic.KafkaListener: topic: topic1
2022-11-12 12:14:11.315  INFO 1023 --- [ner#3-dlt-0-C-1] c.b.b.demo.retrytopic.KafkaListener: topic: topic1-retry-0  
2022-11-12 12:14:12.310  INFO 1023 --- [ner#3-dlt-0-C-1] c.b.b.demo.retrytopic.KafkaListener: topic: topic1-retry-1

2. DLT死信隊列

如果 3 次重試后依舊失敗，會將消息發送到 DLT，

默認情況，消息被發送到死信隊列后，會輸出一條日志。

2022-11-12 12:14:13.324  INFO 1023 --- [ner#3-dlt-0-C-1] o.s.k.retrytopic.RetryTopicConfigurer    : Received message in dlt listener: topic1-dlt@233

DLT的topic為原topic加上后綴“-dlt”

我們可以使用@DltHandler注解來定義進入死信隊列后的操作：

@DltHandler
public void dltHandler(ConsumerRecord< String, String > record) {
    log.info("topic:{}, key:{}, value:{}", record.topic(), record.key(), record.value());
}

3. 自定義@RetryableTopic

可以自定義重試次數、延遲時間、topic命名策略等等，支持使用 Spring EL 表達式讀取配置。

@Slf4j
@Component
public class DemoConsumer {
    @RetryableTopic(
            attempts = "4",
            backoff = @Backoff(delay = "5000", multiplier = "2"),
            fixedDelayTopicStrategy = FixedDelayStrategy.SINGLE_TOPIC
    )
    @KafkaListener(topics = "topic2", groupId = "group1")
    public void onMsg2(ConsumerRecord< String, String > record) {
        log.info("topic: {}", record.topic());
        throw new RuntimeException("kafka exception");
    }

}

注解屬性說明：

attempts ：重試次數，默認為3。

@Backoff delay ：消費延遲時間，單位為毫秒。

@Backoff multiplier ：延遲時間系數，此例中 attempts = 4， delay = 5000， multiplier = 2 ，則間隔時間依次為5s、10s、20s、40s，最大延遲時間受 maxDelay 限制。

fixedDelayTopicStrategy ：可選策略包括：SINGLE_TOPIC 、MULTIPLE_TOPICS

4. 配置類

以上介紹的是注解的方式，只對注解下的方法有效。如果想讓多個方法都用相同的消息重試配置，那么可以使用配置類方式：

@Bean
public RetryTopicConfiguration retryTopic(KafkaTemplate< String, String > template) {
    return RetryTopicConfigurationBuilder
            .newInstance()
            .maxAttempts(4)
            .fixedBackOff(5000)
            .includeTopic("topic1")
            .create(template);
}

小結

以上就是Spring Kafka消息重試機制的簡單應用~希望能夠幫助那些正在使用Spring Kafka或即將使用的人少走一些彎路、少踩一點坑。