咨询Kafka内置重试机制工作原理及相关异常处理、重试次数问题

我来帮你理清这块的细节，毕竟日常用Spring Kafka处理重试场景挺多的，刚好踩过不少坑～

一、异常发生时的具体操作 #

首先得明确：你提到的KafkaListener重试，大多是Spring Kafka框架提供的本地重试能力，和Kafka broker本身的消息重投是两个完全不同的逻辑，我分开给你拆解：

• 本地重试阶段：当你的KafkaListener方法抛出异常后，如果配置了重试（比如用@Retryable注解或者容器的ErrorHandler），不会重新从Kafka broker读取消息——而是直接在当前消费线程中，重复调用你的KafkaListener方法，传入的还是同一条已经拉取到本地内存的ConsumerRecord。

  划重点：这时候消息已经在本地了，重试是纯本地的重复处理，不会和Kafka服务器产生交互（除非重试耗尽后触发offset提交失败）。

• 重试耗尽后的处理：如果重试次数用完还是失败，会根据你的配置走不同逻辑：
  • 若配置了死信队列（DLQ）：框架会把这条消息转发到死信主题，然后正常提交offset，这条消息不会再被原消费者消费。
  • 若没配置死信：会根据你的ack模式决定是否提交offset。比如用MANUAL ack模式且没手动提交，Kafka broker会认为这条消息没被消费，当消费者重新上线或触发Rebalance后，会再次把这条消息推送给消费者——这时候才是“重新从Kafka读取消息”的场景。

二、如何获取当前重试尝试次数 #

根据你用的重试方式不同，获取方式也不一样，给你两种最常用的方案：

1. 使用Spring Retry的@Retryable注解（方法级重试） #

如果是用@Retryable来做方法级的重试，可以通过RetrySynchronizationManager获取当前重试次数：

import org.springframework.retry.annotation.Retryable;
import org.springframework.retry.support.RetrySynchronizationManager;
import org.springframework.kafka.annotation.KafkaListener;

@KafkaListener(topics = "your-topic")
@Retryable(value = {Exception.class}, maxAttempts = 3)
public void listen(ConsumerRecord<String, String> record) {
    // 第一次调用时重试次数为0，每重试一次加1
    int retryCount = RetrySynchronizationManager.getContext().getRetryCount();
    System.out.println("当前重试次数：" + retryCount);
    
    // 你的业务逻辑
    processRecord(record);
}

2. 使用Spring Kafka的DefaultErrorHandler（容器级重试） #

如果是通过容器配置的DefaultErrorHandler来做重试，可以通过消息头获取重试次数：
先在配置类里启用重试次数头：

import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

@Bean
public DefaultErrorHandler errorHandler() {
    // 配置重试3次，每次间隔1秒
    FixedBackOff backOff = new FixedBackOff(1000L, 3);
    DefaultErrorHandler errorHandler = new DefaultErrorHandler(backOff);
    // 启用重试次数头，默认头名称是"kafka_retry_attempt"
    errorHandler.setRetryHeaders(new RetryHeaders());
    return errorHandler;
}

然后在KafkaListener方法中通过@Header注解获取：

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Header;

@KafkaListener(topics = "your-topic")
public void listen(ConsumerRecord<String, String> record,
                   @Header(name = "kafka_retry_attempt", required = false) Integer retryAttempt) {
    // 第一次调用时retryAttempt为null，重试一次后为1，以此类推
    int currentRetryCount = retryAttempt != null ? retryAttempt : 0;
    System.out.println("当前重试次数：" + currentRetryCount);
    
    // 你的业务逻辑
    processRecord(record);
}

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