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

Java 业务端自建 Kafka 重试与死信队列体系 | 极客日志

<!-- Kafka 依赖 -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<!-- 工具类依赖（用于重试次数记录、JSON 序列化） -->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson2</artifactId>
    <version>2.0.32</version>
</dependency>

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 # 偏移量重置策略

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;
}

/**
 * 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";
}

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());
        }
    }
}

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());
        }
    }
}

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());
    }
}

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());
        }
    }
}

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

引言

一、KAFKA 原生重试机制的痛点剖析

二、Java 业务端异常兜底核心方案：自建重试 Topic+ 死信 Topic

核心流程拆解（图文结合理解，建议收藏）

三、方案落地实现（Java 代码实战，直接复制可用）

3.1 环境准备（依赖配置）

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

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

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

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

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

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

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

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

五、结尾总结

更多推荐文章

相关免费在线工具

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

引言

一、KAFKA 原生重试机制的痛点剖析

二、Java 业务端异常兜底核心方案：自建重试 Topic+ 死信 Topic

核心流程拆解（图文结合理解，建议收藏）

三、方案落地实现（Java 代码实战，直接复制可用）

3.1 环境准备（依赖配置）

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

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

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

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

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

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

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

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

五、结尾总结

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具