最近在业务中使用了Spring Kafka，为了更好地掌握其各类用法，我对Spring Kafka进行了系统性的探索，发现了许多实用的特性。接下来，我将介绍一下Spring Kafka的消息重试机制

但让对kafka 稍微了解的都知道，原生 Kafka 并不具备消息重试的功能。不过，从 Spring Kafka 2.7+ 版本开始，引入了封装的 Retry Topic 功能来实现这一需求

@RetryableTopic

可以通过使用注解的方式启用 Retry Topic，只需在 @KafkaListener 方法上添加 @RetryableTopic 注解即可实现

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

这样就启用了 Spring Kafka 的消息重试机制，默认情况下会重试 3 次，每次间隔 1 秒。

我们在方法中模拟了异常抛出，运行后可以看到日志中打印了 3 条记录，间隔约为 1 秒，重试的 topic 是原始 topic 并加上了“-retry”后缀。

2024-08-0212:14:10.230  INFO 1023 --- [ner#3-dlt-0-C-1] c.b.b.demo.retrytopic.KafkaListener: topic: topic12024-08-0212:14:11.315  INFO 1023 --- [ner#3-dlt-0-C-1] c.b.b.demo.retrytopic.KafkaListener: topic: topic1-retry-0  2024-08-0212:14:12.310  INFO 1023 --- [ner#3-dlt-0-C-1] c.b.b.demo.retrytopic.KafkaListener: topic: topic1-retry-1

DLT死信队列

如果重试 3 次后仍然失败，消息将会被发送到死信队列 (DLT)。

默认情况下，当消息被发送到死信队列时，会记录一条日志。

2024-08-0212: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 注解来定义消息进入死信队列后的处理方式。

@DltHandlerpublic void dltHandler(ConsumerRecord record){   log.info("topic:{}, key:{}, value:{}", record.topic(), record.key(), record.value());}

自定义@RetryableTopic

可以自定义重试次数、延迟时间、topic 命名策略等设置，并支持使用 Spring EL 表达式来读取配置。

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

注解属性说明：

attempts：重试次数，默认值为 3。
@Backoff delay：消费延迟时间，单位为毫秒。
@Backoff multiplier：延迟时间的倍增系数。例如，当 attempts = 4、delay = 5000、multiplier = 2 时，间隔时间将依次为 5 秒、10 秒、20 秒、40 秒，最大延迟时间受 maxDelay 限制。
fixedDelayTopicStrategy：可选策略包括 SINGLE_TOPIC 和 MULTIPLE_TOPICS。

配置类

以上介绍的注解方式仅对标注了注解的方法生效。如果希望多个方法都使用相同的消息重试配置，可以采用配置类的方式来实现。

@Beanpublic RetryTopicConfiguration retryTopic(KafkaTemplate template){    return RetryTopicConfigurationBuilder            .newInstance()            .maxAttempts(4)            .fixedBackOff(5000)            .includeTopic("topic1")            .create(template);}

ITPub博客

Kafka的重试机制

最新文章