大家好，我说的是北君。最近在业务中使用了SpringKafka，于是系统地探究了SpringKafka的各种用法，发现了很多实用的特性。下面介绍SpringKafka的消息重试机制。0.PrefaceNativeKafka不支持消息重试。但是SpringKafka2.7+封装了RetryTopic的功能。1、@RetryableTopic使用注解启用RetryTopic，只需在@KafkaListener方法中添加@RetryableTopic即可：String,String>record){log.info("topic:{}",record.topic());thrownewRuntimeException("kafka异常");}}这样就开启了SpringKafka的消息重试机制：默认重试3次，间隔1秒。我们模拟在方法中抛出异常。运行后可以发现打印了3条日志，间隔1秒左右。重试主题为原主题，后缀为“-retry”2022-11-1212:14:10.230INFO1023---[ner#3-dlt-0-C-1]c.b.b.demo.retrytopic.KafkaListener:topic：topic12022-11-1212：14：11.315INFO1023---[ner#3-dlt-0-C-1]c.b.b.demo.retrytopic.KafkaListener：主题：topic1-retry-02022-11-1212：14:12.310INFO1023---[ner#3-dlt-0-C-1]c.b.b.demo.retrytopic.KafkaListener：主题：topic1-retry-12。如果DLT死信队列在3次重试后仍然失败，则将消息发送给DLT。默认情况下，消息发送到死信队列后，会输出一条日志。2022-11-1212:14:13.324INFO1023---[ner#3-dlt-0-C-1]o.s.k.retrytopic.RetryTopicConfigurer：在dlt侦听器中收到消息：topic1-dlt@233DLT的主题已添加到原主题加上“-dlt”后缀，我们可以使用@DltHandler注解来定义进入死信队列后的操作：@DltHandlerpublicvoiddltHandler(ConsumerRecordrecord){log.info("topic:{},key:{},value:{}",record.topic(),record.key(),record.value());}3.自定义@RetryableTopic可以自定义重试次数，延迟时间，topic命名策略等，支持使用SpringEL表达式读取Configuration。@Slf4j@ComponentpublicclassDemoConsumer{@RetryableTopic(attempts="4",backoff=@Backoff(delay="5000",multiplier="2"),fixedDelayTopicStrategy=FixedDelayStrategy.SINGLE_TOPIC)@KafkaListener(topics="topic2",groupId="group1")publicvoidonMsg2(ConsumerRecordrecord){log.info("topic:{}",record.topic());thrownewRuntimeException("kafka异常");}}注解属性说明：attempts：重试次数，默认3次。@Backoffdelay：消费延迟时间，单位毫秒。@Backoffmultiplier：延迟时间系数，本例attempts=4，delay=5000，multiplier=2，则间隔时间为5s，10s，20s，40s，最大延迟时间受maxDelay限制。fixedDelayTopicStrategy：可选策略包括：SINGLE_TOPIC、MULTIPLE_TOPICS4。上面介绍的配置类是注解方法，只对注解下的方法有效。如果想要多个方法重试同一条消息的配置，可以使用配置类方法：@BeanpublicRetryTopicConfigurationretryTopic(KafkaTemplatetemplate){returnRetryTopicConfigurationBuilder.newInstance().maxAttempts(4).fixedBackOff(5000).includeTopic("topic1").create(template);}总结以上就是对SpringKafka消息重试机制的一个简单应用~希望对正在使用SpringKafka或者即将使用它的小伙伴有所帮助少走弯路，少踩坑。
