Java 业务端自建 Kafka 重试与死信队列体系

然后在 application.yml 中配置 Kafka 基本信息（地址、端口、序列化方式等）：

spring:
  kafka:
    bootstrap-servers: 127.0.0.1:9092 # 你的 Kafka 地址
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      retries: 3 # 生产者发送重试（非业务消费重试）
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: dlq-retry-group # 消费组 ID
      enable-auto-commit: false # 关闭自动提交 offset（手动控制）
      auto-offset-reset: earliest # 偏移量重置策略

3.2 核心实体设计（封装消息，记录重试次数）

由于需要记录消息的重试次数，我们需要对原始消息进行封装，新增一个消息载体类：

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;

/**
 * 消息载体（封装原始消息 + 重试次数）
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public class KafkaRetryMessage implements Serializable {
    // 原始消息内容（JSON 格式）
    private String originalMessage;
    // 当前重试次数（初始为 0）
    private Integer retryCount;
    // 消息唯一标识（用于去重，可选）
    private String messageId;
    // 首次消费时间（用于排查问题）
    private Long firstConsumeTime;
}

3.3 Topic 常量定义（统一管理，避免硬编码）

定义 3 类 Topic 的名称，后续新增、修改时只需修改常量，无需改动业务代码：

/**
 * Kafka Topic 常量类
 */
public class KafkaTopicConstant {
    // 业务 Topic（示例：用户下单消息）
    public static final String BUSINESS_TOPIC = "user-order-topic";
    // 重试 Topic（按重试次数分级，这里简化为 1 个，可扩展为 retry-topic-1、retry-topic-2）
    public static final String RETRY_TOPIC = "retry-topic";
    // 死信 Topic
    public static final String DLQ_TOPIC = "dlq-topic";
}

3.4 消费者实现（业务消费 + 重试消费 + 死信消费）

3.4.1 业务消费者（监听业务 Topic，处理核心逻辑）

核心逻辑：消费业务消息，执行业务逻辑，捕获异常后判断重试次数，发送到重试 Topic 或死信 Topic。

import com.alibaba.fastjson2.JSON;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;

@Component
public class BusinessConsumer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    
    // 重试次数阈值（可配置在 yml 中，这里简化为常量）
    private static final Integer RETRY_MAX_COUNT = 3;

    /**
     * 监听业务 Topic，处理核心业务逻辑
     */
    @KafkaListener(topics = KafkaTopicConstant.BUSINESS_TOPIC, groupId = "${spring.kafka.consumer.group-id}")
    public void consumeBusinessMessage(String message) {
        try {
            // 1. 解析消息（这里假设原始消息是 JSON 格式，封装为重试消息载体）
            KafkaRetryMessage retryMessage = JSON.parseObject(message, KafkaRetryMessage.class);
            // 2. 执行核心业务逻辑（示例：用户下单处理）
            handleOrderBusiness(retryMessage.getOriginalMessage());
            // 3. 消费成功，手动提交 offset（由 Spring Kafka 自动管理，无需手动调用）
            System.out.println("消息消费成功，messageId：" + retryMessage.getMessageId());
        } catch (Exception e) {
            // 4. 消费失败，处理重试逻辑
            handleConsumeFail(message, e);
        }
    }

    /**
     * 核心业务逻辑（示例：用户下单）
     */
    private void handleOrderBusiness(String originalMessage) {
        // 这里模拟业务异常（如数据库连接超时、接口调用失败）
        // 实际开发中替换为真实业务逻辑（如调用订单服务、库存服务）
        // throw new RuntimeException("数据库连接超时，消费失败");
    }

    /**
     * 消费失败处理：判断重试次数，发送到重试 Topic 或死信 Topic
     */
    private void handleConsumeFail(String message, Exception e) {
        KafkaRetryMessage retryMessage = JSON.parseObject(message, KafkaRetryMessage.class);
        Integer currentRetryCount = retryMessage.getRetryCount();
        System.out.println("消息消费失败，messageId：" + retryMessage.getMessageId() + "，当前重试次数：" + currentRetryCount + "，异常信息：" + e.getMessage());
        
        // 判断是否达到重试阈值
        if (currentRetryCount < RETRY_MAX_COUNT) {
            // 未达阈值：重试次数 +1，发送到重试 Topic
            retryMessage.setRetryCount(currentRetryCount + 1);
            kafkaTemplate.send(KafkaTopicConstant.RETRY_TOPIC, JSON.toJSONString(retryMessage));
            System.out.println("消息已发送到重试 Topic，下次重试次数：" + (currentRetryCount + 1));
        } else {
            // 已达阈值：发送到死信 Topic，结束重试
            kafkaTemplate.send(KafkaTopicConstant.DLQ_TOPIC, JSON.toJSONString(retryMessage));
            System.out.println("消息重试次数已达阈值，发送到死信 Topic，messageId：" + retryMessage.getMessageId());
        }
    }
}

3.4.2 重试消费者（监听重试 Topic，实现延迟重试）

这里的关键是'延迟重试'——避免频繁重试，我们可以通过'消费者拉取间隔'或'消息延迟发送'实现，这里采用简单易落地的'拉取间隔配置'：

import com.alibaba.fastjson2.JSON;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;

@Component
public class RetryConsumer {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    
    private static final Integer RETRY_MAX_COUNT = 3;

    /**
     * 监听重试 Topic，实现延迟重试
     * 注：通过 concurrency 控制消费者线程数，通过 poll-timeout 控制拉取间隔（实现延迟）
     */
    @KafkaListener(
        topics = KafkaTopicConstant.RETRY_TOPIC,
        groupId = "${spring.kafka.consumer.group-id}",
        concurrency = "1", // 单线程，避免并发重试导致的问题
        properties = {"max.poll.records=10", "poll.timeout.ms=5000"} // 拉取间隔 5 秒，实现延迟重试
    )
    public void consumeRetryMessage(String message) {
        try {
            KafkaRetryMessage retryMessage = JSON.parseObject(message, KafkaRetryMessage.class);
            // 重新执行业务逻辑（与业务消费者逻辑一致，可抽取为公共方法）
            handleOrderBusiness(retryMessage.getOriginalMessage());
            System.out.println("重试消息消费成功，messageId：" + retryMessage.getMessageId() + "，重试次数：" + retryMessage.getRetryCount());
        } catch (Exception e) {
            // 重试消费失败，再次判断重试次数
            handleConsumeFail(message, e);
        }
    }

    // 复用业务逻辑方法（实际开发中可抽取到 Service 层）
    private void handleOrderBusiness(String originalMessage) {
        // 与业务消费者的 handleOrderBusiness 方法一致
        // throw new RuntimeException("重试消费失败，模拟异常");
    }

    // 复用消费失败处理方法（实际开发中可抽取为公共工具类）
    private void handleConsumeFail(String message, Exception e) {
        KafkaRetryMessage retryMessage = JSON.parseObject(message, KafkaRetryMessage.class);
        Integer currentRetryCount = retryMessage.getRetryCount();
        if (currentRetryCount < RETRY_MAX_COUNT) {
            retryMessage.setRetryCount(currentRetryCount + 1);
            kafkaTemplate.send(KafkaTopicConstant.RETRY_TOPIC, JSON.toJSONString(retryMessage));
            System.out.println("重试消息再次失败，继续发送到重试 Topic，下次重试次数：" + (currentRetryCount + 1));
        } else {
            kafkaTemplate.send(KafkaTopicConstant.DLQ_TOPIC, JSON.toJSONString(retryMessage));
            System.out.println("重试消息已达最大次数，发送到死信 Topic，messageId：" + retryMessage.getMessageId());
        }
    }
}

3.4.3 死信消费者（监听死信 Topic，兜底处理）

死信消息的核心作用是'兜底'，避免消息丢失，同时方便排查问题，因此我们需要将死信消息持久化（这里模拟存入数据库，实际开发中可根据需求调整）：

import com.alibaba.fastjson2.JSON;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class DlqConsumer {
    /**
     * 监听死信 Topic，处理最终失败的消息
     */
    @KafkaListener(topics = KafkaTopicConstant.DLQ_TOPIC, groupId = "${spring.kafka.consumer.group-id}")
    public void consumeDlqMessage(String message) {
        KafkaRetryMessage retryMessage = JSON.parseObject(message, KafkaRetryMessage.class);
        System.out.println("接收死信消息，messageId：" + retryMessage.getMessageId() + "，原始消息：" + retryMessage.getOriginalMessage());
        
        // 核心操作：将死信消息持久化（存入数据库、ES 等），供人工排查
        // 这里模拟持久化操作
        saveDlqMessageToDb(retryMessage);
        
        // 可选：发送告警通知（如钉钉、企业微信），提醒开发人员处理
        sendDlqAlarm(retryMessage);
    }

    /**
     * 死信消息持久化到数据库
     */
    private void saveDlqMessageToDb(KafkaRetryMessage retryMessage) {
        // 实际开发中，调用 DAO 层方法，将消息存入数据库（如 dlq_message 表）
        System.out.println("死信消息已持久化，messageId：" + retryMessage.getMessageId());
    }

    /**
     * 发送死信告警通知
     */
    private void sendDlqAlarm(KafkaRetryMessage retryMessage) {
        // 调用告警工具类，发送钉钉/企业微信通知
        System.out.println("已发送死信告警，提醒处理 messageId：" + retryMessage.getMessageId());
    }
}

3.5 生产者测试（模拟消息发送，验证流程）

编写一个测试类，模拟发送业务消息，验证整个'业务消费 - 重试 - 死信'流程：

import com.alibaba.fastjson2.JSON;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import javax.annotation.Resource;
import java.util.UUID;

@SpringBootTest
public class KafkaRetryDlqTest {
    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @Test
    public void sendBusinessMessage() {
        // 模拟发送 10 条业务消息，故意让其消费失败（打开业务逻辑中的异常抛出）
        for (int i = 0; i < 10; i++) {
            KafkaRetryMessage retryMessage = new KafkaRetryMessage();
            retryMessage.setOriginalMessage("{\"orderId\":\"ORDER_" + i + "\",\"userId\":\"USER_100" + i + "\",\"amount\":100.0}");
            retryMessage.setRetryCount(0); // 初始重试次数为 0
            retryMessage.setMessageId(UUID.randomUUID().toString());
            retryMessage.setFirstConsumeTime(System.currentTimeMillis());
            
            // 发送到业务 Topic
            kafkaTemplate.send(KafkaTopicConstant.BUSINESS_TOPIC, JSON.toJSONString(retryMessage));
            System.out.println("发送业务消息成功，messageId：" + retryMessage.getMessageId());
        }
    }
}

四、关键注意事项与避坑指南（必看！）

1. 重试次数与延迟策略：重试次数建议设置为 3-5 次（过多会导致消息积压），延迟策略可根据业务调整（如首次延迟 5 秒、第二次延迟 10 秒、第三次延迟 30 秒，可通过多 Retry Topic 实现分级延迟）；
2. 消息去重：由于重试机制的存在，可能会出现消息重复消费的情况，建议在业务层实现幂等性（如通过 messageId 去重、数据库唯一约束）；
3. Topic 分区设计：重试 Topic 和死信 Topic 的分区数，建议与业务 Topic 一致，避免分区不均衡导致的消费延迟；
4. 监控告警：务必给死信队列添加监控和告警（如消息积压监控、告警通知），否则死信消息堆积后无法及时发现；
5. 资源控制：重试消费者的线程数不宜过多，避免频繁重试占用过多系统资源，可根据业务并发量调整；
6. 序列化方式：建议使用 JSON 或 Protobuf 序列化消息，避免使用 Java 原生序列化（易出现兼容性问题）；
7. offset 提交：必须关闭自动提交 offset，采用手动提交（Spring Kafka 可通过@KafkaListener 的 ackMode 配置实现），避免消费失败后 offset 被提交，导致消息丢失。

五、结尾总结

本文围绕 Kafka 原生重试机制的痛点，详细讲解了 Java 业务端如何通过'自建重试 Topic+ 死信 Topic'，打造闭环的消息异常容错体系。核心是将重试逻辑从 Kafka 原生机制中剥离，实现可控的重试次数、延迟策略，同时通过死信队列兜底，确保消息不丢失、系统高可用。

整个方案的优势在于：无需修改 Kafka 服务器配置，完全在业务端实现，开发成本低、可扩展性强，适合各类 Java 业务场景（如电商、支付、日志处理等）。