如何用Spring Kafka标识生产者端的重试消息？

针对你遇到的问题——需要让消费者区分生产者首次发送和重试发送的消息，这里提供几个可行的实现方案：

方案1：通过自定义生产者拦截器添加重试次数消息头 #

这是侵入性最小、最可靠的方案，利用Kafka的ProducerInterceptor在每次发送（包括重试）时给消息添加重试次数的头信息，消费者通过读取这个头来判断是否为重试消息。

实现步骤： #

1. 编写自定义拦截器类

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import java.util.Map;

public class RetryCountInterceptor implements ProducerInterceptor<String, String> {

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
        // 读取当前消息的重试次数，首次发送默认0
        Integer retryCount = record.headers().lastHeader("x-retry-count") != null 
            ? Integer.parseInt(new String(record.headers().lastHeader("x-retry-count").value()))
            : 0;
        // 每次发送（含重试）次数+1
        retryCount++;
        // 移除旧的头字段，添加更新后的重试次数
        record.headers().remove("x-retry-count");
        record.headers().add("x-retry-count", String.valueOf(retryCount).getBytes());
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {}

    @Override
    public void close() {}

    @Override
    public void configure(Map<String, ?> configs) {}
}

2. 在生产者配置中添加拦截器

Properties props = new Properties();
// 保留你原有的重试配置
props.put(RETRIES_CONFIG, 20);
props.put(RETRY_BACKOFF_MS_CONFIG, "500");
props.put(DELIVERY_TIMEOUT_MS_CONFIG, "5000");
// 基础配置
props.put(BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONTAINER.getBootstrapServers());
props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

// 注册自定义拦截器
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, RetryCountInterceptor.class.getName());

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

3. 消费者端读取头信息做判断

@KafkaListener(topics = "your-target-topic")
public void handleMessage(ConsumerRecord<String, String> record) {
    Header retryHeader = record.headers().lastHeader("x-retry-count");
    if (retryHeader != null) {
        int retryCount = Integer.parseInt(new String(retryHeader.value()));
        if (retryCount > 1) {
            // 这是经过重试的消息，执行你的特殊处理逻辑
            System.out.println("处理重试消息，重试次数：" + retryCount);
        } else {
            // 首次发送的消息，正常处理
        }
    }
}

方案2：在消息体中嵌入重试标记（适合可修改消息结构的场景） #

如果你的消息结构可以调整，可以把消息封装成包含重试次数的DTO，生产者在发送时设置初始值，重试时更新该值。

示例DTO：

public class BusinessMessage {
    private String content;
    private int retryCount;

    // 构造器、getter、setter省略
}

这种方式需要消费者同步解析该DTO，适合内部系统的消息交互场景。

方案3：基于时间差的间接判断（不推荐） #

Kafka本身不会在消息中内置重试元数据，你可以尝试通过消息的发送时间戳和消费者接收时间的差值，结合重试间隔来间接判断，但这种方式受网络延迟、集群负载等因素影响很大，准确性无法保证，仅作为备选方案。

注意点 #

• 拦截器会在每次发送请求时触发，包括生产者自动重试的请求，所以能准确统计重试次数
• 消息头字段不会修改消息体内容，对现有业务逻辑的侵入性极低
• 如果使用Spring Kafka的KafkaTemplate，只需要在ProducerFactory的配置中添加拦截器属性即可，逻辑一致

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