前言：为什么要自研消息队列？

在绝大多数Java后端业务场景中，我们习惯性直接使用RocketMQ、Kafka、RabbitMQ等成熟开源消息中间件。这些组件经过多年迭代，性能、稳定性、可用性都经过了大厂海量流量打磨，足以支撑绝大多数企业级业务。但在我上次负责的高并发电商秒杀、订单异步核销、日志实时投递业务场景中，团队最终选择放弃开源MQ，基于Java原生技术栈自研一套轻量级百万级消息队列。

核心原因有三点：

1. 业务轻量化诉求：团队业务无需分布式集群、跨机器投递能力，仅需服务内高吞吐异步解耦，部署开源MQ存在资源冗余、运维成本过高的问题；

2. 极致性能可控：开源MQ存在网络IO、协议解析、集群同步的固有损耗，自研可基于业务场景裁剪逻辑，最大化提升单机吞吐；

3. 定制化能力刚需：业务需要绑定本地事务、自定义消息优先级、毫秒级延时投递，开源MQ的适配改造成本远高于自研。

初期设计目标很简单：单机支撑百万级消息吞吐、无消息丢失、支持异步消费、线程安全、低内存占用。开发阶段一切顺利，基于Java阻塞队列、线程池、本地持久化快速完成开发，压测环境完美达标，单机峰值吞吐可达120万+/分钟。

但真正上线生产环境后，各种隐藏深坑接连爆发：消息无声丢失、消费堆积雪崩、内存泄漏、线程死锁、重复消费、持久化损坏、超时积压连锁故障。这些问题在测试环境完全无法复现，仅在高并发、长时间运行、网络波动、服务重启的生产场景集中爆发。

本文将完整复盘本次自研百万级消息队列的架构设计、核心代码、线上故障复盘、坑点原理、落地优化方案，所有问题均为生产真实踩坑，所有代码均可直接复用，帮大家避开自研队列的致命误区。全文万字干货，覆盖自研MQ从入门到避坑的全流程。

一、自研百万级消息队列初始架构设计

1.1 核心设计思路

本次自研MQ定位为单机内存级消息队列+本地磁盘持久化，无分布式架构，专注服务内高并发异步解耦，核心架构分为三大模块：

- 生产者模块：接收业务消息，支持同步/异步投递、消息优先级划分、消息预处理；

- 队列存储模块：内存缓冲队列+磁盘落底机制，高并发走内存提升吞吐，空闲时段异步持久化防止消息丢失；

- 消费者模块：独立线程池轮询消费，支持批量消费、失败重试、异常兜底。

初始技术选型全部基于Java原生API，无第三方依赖，保证轻量化、高性能：

- 内存队列：采用JUC包下ArrayBlockingQueue，基于数组实现，有界阻塞队列，线程安全、吞吐稳定；

- 消费线程池：自定义ThreadPoolExecutor，固定核心线程数，避免线程频繁创建销毁损耗；

- 持久化机制：Java IO随机读写流，本地磁盘追加写入消息日志；

- 消息模型：自定义消息实体，包含消息ID、业务类型、消息体、时间戳、重试次数、优先级等核心字段。

1.2 初始核心代码实现

1.2.1 自定义消息实体类

承载所有业务消息数据，记录消息生命周期核心参数，为后续重试、幂等、溯源提供基础字段。

import java.io.Serializable;

/**
 * 自研MQ自定义消息实体
 */
public class MqMessage implements Serializable {
    // 全局唯一消息ID（初始用UUID生成）
    private String messageId;
    // 业务消息类型（订单、日志、核销等）
    private String bizType;
    // 消息主体内容
    private String content;
    // 消息创建时间戳
    private long createTime;
    // 消息重试次数
    private int retryCount;
    // 消息优先级 1-10，数值越大优先级越高
    private int priority;
    // 消息状态 0-待消费 1-消费成功 2-消费失败 3-死信
    private int status;

    // 无参、有参构造、get/set方法
    public MqMessage() {}

    public MqMessage(String messageId, String bizType, String content, int priority) {
        this.messageId = messageId;
        this.bizType = bizType;
        this.content = content;
        this.createTime = System.currentTimeMillis();
        this.retryCount = 0;
        this.priority = priority;
        this.status = 0;
    }

    // 省略getter/setter
    public String getMessageId() { return messageId; }
    public void setMessageId(String messageId) { this.messageId = messageId; }
    public String getBizType() { return bizType; }
    public void setBizType(String bizType) { this.bizType = bizType; }
    public String getContent() { return content; }
    public void setContent(String content) { this.content = content; }
    public long getCreateTime() { return createTime; }
    public void setCreateTime(long createTime) { this.createTime = createTime; }
    public int getRetryCount() { return retryCount; }
    public void setRetryCount(int retryCount) { this.retryCount = retryCount; }
    public int getPriority() { return priority; }
    public void setPriority(int priority) { this.priority = priority; }
    public int getStatus() { return status; }
    public void setStatus(int status) { this.status = status; }
}

1.2.2 队列核心存储管理器

初始化内存阻塞队列，定义消息投递、获取核心方法，对接本地持久化逻辑，是整个MQ的核心载体。

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * 消息队列核心管理器
 */
public class MqQueueManager {
    // 定义队列最大容量，初始设置100万容量，支撑百万级消息缓冲
    private static final int MAX_QUEUE_SIZE = 1000000;
    // 内存阻塞队列
    private final BlockingQueue<MqMessage> messageQueue;
    // 持久化工具类
    private final MqPersistence persistence;

    // 单例初始化
    private static final MqQueueManager INSTANCE = new MqQueueManager();

    private MqQueueManager() {
        // 初始化有界阻塞队列
        messageQueue = new ArrayBlockingQueue<>(MAX_QUEUE_SIZE);
        // 初始化本地持久化
        persistence = new MqPersistence();
        // 服务启动加载磁盘未消费消息
        persistence.loadMessageFromDisk(messageQueue);
    }

    public static MqQueueManager getInstance() {
        return INSTANCE;
    }

    /**
     * 投递消息到队列
     */
    public boolean pushMessage(MqMessage message) {
        if (message == null) {
            return false;
        }
        // 内存入队
        boolean offer = messageQueue.offer(message);
        if (offer) {
            // 异步持久化到磁盘
            persistence.asyncSaveMessage(message);
        }
        return offer;
    }

    /**
     * 阻塞获取消息，用于消费者轮询
     */
    public MqMessage takeMessage() throws InterruptedException {
        return messageQueue.take();
    }

    /**
     * 获取当前队列积压消息数
     */
    public int getQueueSize() {
        return messageQueue.size();
    }
}

1.2.3 本地持久化工具类

通过文件追加写入实现消息落盘，服务重启后加载磁盘消息，初始设计用于解决服务重启消息丢失问题。

import com.alibaba.fastjson.JSON;
import java.io.*;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.concurrent.BlockingQueue;

/**
 * 消息本地磁盘持久化工具
 */
public class MqPersistence {
    // 消息持久化存储路径
    private static final String MESSAGE_FILE_PATH = "./mq_message.log";
    // 文件写入流
    private BufferedWriter writer;

    public MqPersistence() {
        try {
            // 初始化文件写入流，追加模式
            File file = new File(MESSAGE_FILE_PATH);
            if (!file.exists()) {
                file.createNewFile();
            }
            writer = Files.newBufferedWriter(Paths.get(MESSAGE_FILE_PATH), 
                    java.nio.charset.StandardCharsets.UTF_8, 
                    java.nio.file.StandardOpenOption.APPEND);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 异步持久化消息
     */
    public void asyncSaveMessage(MqMessage message) {
        // 简单异步线程落盘
        new Thread(() -> {
            try {
                // JSON序列化写入文件，每行一条消息
                writer.write(JSON.toJSONString(message));
                writer.newLine();
                writer.flush();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
    }

    /**
     * 服务启动加载磁盘消息到内存队列
     */
    public void loadMessageFromDisk(BlockingQueue<MqMessage> queue) {
        File file = new File(MESSAGE_FILE_PATH);
        if (!file.exists() || file.length() == 0) {
            return;
        }
        try (BufferedReader reader = Files.newBufferedReader(Paths.get(MESSAGE_FILE_PATH))) {
            String line;
            while ((line = reader.readLine()) != null) {
                MqMessage message = JSON.parseObject(line, MqMessage.class);
                queue.offer(message);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

1.2.4 消费者线程核心逻辑

自定义消费线程池，循环从队列获取消息并执行业务消费逻辑，支持失败重试。

import java.util.concurrent.*;

/**
 * 消息消费者核心处理器
 */
public class MqConsumer {
    // 消费核心线程数，初始固定10线程
    private static final int CONSUMER_THREAD_NUM = 10;
    private final ThreadPoolExecutor consumerThreadPool;
    private final MqQueueManager queueManager;
    // 最大重试次数
    private static final int MAX_RETRY_NUM = 3;

    public MqConsumer() {
        this.queueManager = MqQueueManager.getInstance();
        // 初始化消费线程池
        this.consumerThreadPool = new ThreadPoolExecutor(
                CONSUMER_THREAD_NUM,
                CONSUMER_THREAD_NUM,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                r -> new Thread(r, "mq-consumer-thread-" + r.hashCode())
        );
        // 启动消费任务
        startConsume();
    }

    /**
     * 启动循环消费
     */
    private void startConsume() {
        // 固定线程循环消费
        for (int i = 0; i < CONSUMER_THREAD_NUM; i++) {
            consumerThreadPool.execute(this::consumeLoop);
        }
    }

    /**
     * 消费死循环
     */
    private void consumeLoop() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                // 阻塞获取消息
                MqMessage message = queueManager.takeMessage();
                // 执行业务消费逻辑
                boolean consumeSuccess = doConsume(message);
                if (!consumeSuccess) {
                    // 消费失败，重试
                    retryMessage(message);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 执行业务消费（模拟业务逻辑）
     */
    private boolean doConsume(MqMessage message) {
        try {
            // 模拟不同业务消费逻辑：订单处理、日志上报、数据统计等
            System.out.println("消费消息：" + message.getMessageId() + "，内容：" + message.getContent());
            // 模拟业务耗时
            TimeUnit.MILLISECONDS.sleep(20);
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    /**
     * 消息重试机制
     */
    private void retryMessage(MqMessage message) {
        if (message.getRetryCount() < MAX_RETRY_NUM) {
            message.setRetryCount(message.getRetryCount() + 1);
            // 重新投递队列重试
            queueManager.pushMessage(message);
        } else {
            // 超过最大重试，标记死信
            message.setStatus(3);
            System.err.println("消息重试耗尽，进入死信：" + message.getMessageId());
        }
    }
}

1.3 压测环境完美达标

开发完成后，我们基于JMeter做了高压压测：单机模拟10个生产者线程，每秒投递2000条消息，持续压测1小时。

压测结果：

1. 峰值吞吐：124万消息/分钟，达到百万级设计目标；

2. 无消息丢失、无异常报错；

3. 内存占用稳定，CPU使用率维持在40%以内；

4. 服务重启后可正常加载磁盘消息，无数据丢失。

基于压测结果，团队判定队列满足生产要求，直接打包上线。谁也没想到，压测的完美数据，恰恰掩盖了所有隐藏坑点。上线72小时后，线上故障全面爆发。

二、上线后爆发的10大致命隐藏坑（附故障现场+原理+修复）

自研MQ的坑和开源MQ完全不同，开源MQ的问题多为配置不当、集群问题，而自研MQ的问题全部是架构设计缺陷、代码细节漏洞、并发模型错误、边界场景缺失导致，且只在生产高并发、长时间运行、极端边界场景触发，测试环境100%无法复现。

坑点1：异步持久化多线程竞争，导致消息丢失+文件损坏

2.1.1 线上故障现象

上线24小时后，运维监控发现：业务日志显示消息投递成功，但部分订单消息完全没有消费记录，服务重启后丢失消息数量大幅增加；同时偶尔出现mq_message.log文件内容错乱、半截数据、空行，导致启动加载时报错，批量丢失历史消息。

2.1.2 问题根因分析

回看持久化代码，核心漏洞极其隐蔽：全局唯一BufferedWriter多线程并发写入，无锁控制。

我们的asyncSaveMessage方法，每次投递消息都会新建一个线程执行写入操作，所有线程共用同一个全局BufferedWriter。而BufferedWriter是非线程安全的，多线程并发write、flush时，会出现：

1. 数据覆盖：多个线程的消息内容互相覆盖，导致单条消息残缺；

2. 数据穿插：两条消息内容拼接在同一行，JSON解析失败，启动加载直接丢弃；

3. 缓冲区刷新异常：部分数据滞留缓冲区，未落地磁盘，服务重启彻底丢失。

压测环境之所以无问题，是因为压测消息均匀、并发规律，极少触发线程竞争临界条件，而生产环境消息突发、峰值并发极高，竞争概率100%触发。

2.1.3 完整修复方案

1. 加入对象锁，保证文件写入串行化；

2. 去掉频繁新建线程的低效逻辑，改用线程池统一执行持久化任务；

3. 增加消息写入校验、文件修复兜底逻辑；

4. 批量刷盘，减少IO频繁操作，提升性能。

import com.alibaba.fastjson.JSON;
import java.io.*;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 修复后持久化工具类（解决多线程写入错乱、消息丢失）
 */
public class MqPersistence {
    private static final String MESSAGE_FILE_PATH = "./mq_message.log";
    private BufferedWriter writer;
    // 新增：写入锁，保证线程安全
    private final Object WRITE_LOCK = new Object();
    // 新增：专用持久化线程池
    private final ScheduledExecutorService persistenceExecutor;
    // 批量刷盘阈值
    private static final int BATCH_FLUSH_SIZE = 50;
    private int cacheCount = 0;

    public MqPersistence() {
        persistenceExecutor = Executors.newSingleThreadScheduledExecutor();
        try {
            File file = new File(MESSAGE_FILE_PATH);
            if (!file.exists()) {
                file.createNewFile();
            }
            writer = Files.newBufferedWriter(Paths.get(MESSAGE_FILE_PATH),
                    java.nio.charset.StandardCharsets.UTF_8,
                    java.nio.file.StandardOpenOption.APPEND);
            // 定时强制刷盘，避免缓冲区数据滞留
            persistenceExecutor.scheduleAtFixedRate(this::forceFlush, 1, 1, TimeUnit.SECONDS);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 修复后：线程安全异步持久化
     */
    public void asyncSaveMessage(MqMessage message) {
        persistenceExecutor.execute(() -> {
            synchronized (WRITE_LOCK) {
                try {
                    String jsonStr = JSON.toJSONString(message);
                    writer.write(jsonStr);
                    writer.newLine();
                    cacheCount++;
                    // 达到阈值批量刷盘
                    if (cacheCount >= BATCH_FLUSH_SIZE) {
                        forceFlush();
                        cacheCount = 0;
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    /**
     * 强制刷盘落地
     */
    private void forceFlush() {
        synchronized (WRITE_LOCK) {
            try {
                if (writer != null) {
                    writer.flush();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 优化加载：过滤残缺异常消息
     */
    public void loadMessageFromDisk(BlockingQueue<MqMessage> queue) {
        File file = new File(MESSAGE_FILE_PATH);
        if (!file.exists() || file.length() == 0) {
            return;
        }
        try (BufferedReader reader = Files.newBufferedReader(Paths.get(MESSAGE_FILE_PATH))) {
            String line;
            while ((line = reader.readLine()) != null) {
                try {
                    // 过滤空行、残缺数据
                    if (line.trim().isEmpty()) continue;
                    MqMessage message = JSON.parseObject(line, MqMessage.class);
                    queue.offer(message);
                } catch (Exception e) {
                    // 记录损坏消息日志，便于溯源
                    System.err.println("解析损坏消息数据：" + line);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

坑点2：ArrayBlockingQueue队列满后消息静默丢弃，无任何告警

2.2.1 线上故障现象

上线36小时，电商大促峰值时段，大量用户反馈下单成功但订单未生成、支付后无核销记录。业务日志显示生产者执行pushMessage返回true，但队列积压数量不涨，大量消息直接消失，无报错、无异常，完全静默丢失。

2.2.2 问题根因分析

初始队列采用ArrayBlockingQueue.offer()方法投递消息，这是核心致命误区。

offer()方法特性：队列已满时，直接返回false，无异常、无阻塞、无日志，静默丢弃元素。

我们初始设置队列最大容量100万，压测匀速投递不会打满队列，但生产大促峰值是瞬时脉冲流量，1秒内涌入数十万消息，消费线程处理速度跟不上生产速度，瞬间打满队列。此时后续所有消息offer直接返回false，业务无任何感知，导致核心业务消息批量丢失。

更致命的是：初始代码仅判断返回值，未做任何告警、降级、阻塞兜底，相当于给业务埋了一颗隐形炸弹。

2.2.3 完整修复方案

1. 替换投递策略：核心业务消息使用put()阻塞投递，非核心消息使用带超时的offer；

2. 增加队列阈值监控，临近容量上限触发告警；

3. 队列满时新增本地临时缓存兜底，避免瞬时流量丢消息；

4. 完善日志埋点，记录投递成功、失败、队列满场景日志。

/**
 * 修复后消息投递方法
 */
// 新增：队列告警阈值（80%容量触发告警）
private static final int QUEUE_ALERT_THRESHOLD = (int) (MAX_QUEUE_SIZE * 0.8);

public boolean pushMessage(MqMessage message) throws InterruptedException {
    if (message == null) {
        return false;
    }
    // 队列容量告警判断
    int currentSize = messageQueue.size();
    if (currentSize > QUEUE_ALERT_THRESHOLD) {
        // 触发告警（可对接钉钉/邮件告警）
        System.err.println("【队列告警】当前队列积压量过高：" + currentSize + "，阈值：" + QUEUE_ALERT_THRESHOLD);
    }

    boolean pushResult;
    // 核心业务消息阻塞投递，避免丢失
    if ("ORDER".equals(message.getBizType()) || "PAY".equals(message.getBizType())) {
        messageQueue.put(message);
        pushResult = true;
    } else {
        // 非核心消息3秒超时投递，避免长时间阻塞业务线程
        pushResult = messageQueue.offer(message, 3, TimeUnit.SECONDS);
        if (!pushResult) {
            // 超时失败，记录 error 日志，便于排查
            System.err.println("【消息投递失败】队列已满，非核心消息丢弃：" + message.getMessageId());
        }
    }

    if (pushResult) {
        persistence.asyncSaveMessage(message);
    }
    return pushResult;
}

坑点3：消费线程池固定线程数，引发大流量堆积+雪崩

2.3.1 线上故障现象

每次流量峰值过后，队列消息积压量持续飙升，从几万堆积到百万级别，消费速度极慢，甚至出现消费停滞。重启服务后瞬间清空部分堆积，但新一轮流量过来再次堆积，最终导致业务异步逻辑严重滞后，数据统计、订单核销延迟数分钟。

2.3.2 问题根因分析

初始消费线程池采用固定线程池，核心线程、最大线程均为10，无扩容能力。该配置存在致命缺陷：

1. 业务消费逻辑存在偶尔耗时波动（比如数据库查询超时、外部接口延迟），单个线程阻塞会导致消费能力下降；

2. 固定线程数无法应对流量峰值，生产速度 > 消费速度，消息持续堆积；

3. 线程池队列无限制，阻塞任务堆积，占用大量内存，导致GC频繁，进一步拖慢消费效率，形成堆积-GC-更堆积的雪崩循环。

结合行业数据，消费端问题占MQ线上故障的80%，线程池配置不合理是首要诱因。

2.3.3 完整修复方案

1. 替换为动态可扩容线程池，适配流量波动；

2. 增加线程池超时销毁、空闲回收机制；

3. 新增批量消费逻辑，提升吞吐；

4. 增加堆积监控，动态调整线程数。

/**
 * 修复后动态消费线程池+批量消费
 */
public class MqConsumer {
    // 核心线程数、最大线程数扩容
    private static final int CORE_THREAD_NUM = 10;
    private static final int MAX_THREAD_NUM = 30;
    private static final long KEEP_ALIVE_TIME = 60L;
    // 批量消费条数
    private static final int BATCH_CONSUME_SIZE = 20;
    private final ThreadPoolExecutor consumerThreadPool;
    private final MqQueueManager queueManager;
    private static final int MAX_RETRY_NUM = 3;

    public MqConsumer() {
        this.queueManager = MqQueueManager.getInstance();
        // 动态线程池：可扩容、空闲线程自动回收
        this.consumerThreadPool = new ThreadPoolExecutor(
                CORE_THREAD_NUM,
                MAX_THREAD_NUM,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new SynchronousQueue<>(), // 无队列缓冲，任务直接创建线程
                r -> new Thread(r, "mq-consumer-thread-" + r.hashCode()),
                new ThreadPoolExecutor.CallerRunsPolicy() // 队列满时主线程执行，避免丢弃
        );
        startBatchConsume();
    }

    /**
     * 启动批量消费任务
     */
    private void startBatchConsume() {
        for (int i = 0; i < CORE_THREAD_NUM; i++) {
            consumerThreadPool.execute(this::batchConsumeLoop);
        }
    }

    /**
     * 批量消费循环，大幅提升吞吐
     */
    private void batchConsumeLoop() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                // 批量获取消息
                List<MqMessage> messageList = new ArrayList<>(BATCH_CONSUME_SIZE);
                queueManager.getMessageQueue().drainTo(messageList, BATCH_CONSUME_SIZE);
                if (messageList.isEmpty()) {
                    // 无消息短暂休眠，避免空转消耗CPU
                    TimeUnit.MILLISECONDS.sleep(10);
                    continue;
                }
                // 批量消费
                for (MqMessage message : messageList) {
                    boolean success = doConsume(message);
                    if (!success) {
                        retryMessage(message);
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

坑点4：无幂等机制，高并发下重复消费引发业务事故

2.4.1 线上故障现象

上线48小时，出现多起严重业务事故：同一笔订单多次扣款、用户积分重复发放、订单状态重复更新、数据库重复插入多条相同数据。经排查，均为同一条消息被多次消费导致。

2.4.2 问题根因分析

自研队列初始完全忽略了消息重试、线程重试、服务重启重试带来的重复消费问题。

触发重复消费的三大场景：

1. 消息消费过程中业务超时，被判定为消费失败，触发重试机制；

2. 消费线程执行中被中断、服务重启，未标记消费成功的消息重新入队；

3. 瞬时流量下消息投递重复、重试逻辑叠加，导致同消息多次消费。

所有MQ的通用铁律：生产环境中，重复消费是必然事件，绝对不存在严格的一次性消费，必须业务层做幂等。开源MQ默认保证至少一次投递，自研队列更是无法规避该问题。

2.4.3 完整修复方案

采用「全局消息ID+本地缓存+数据库唯一索引」三重幂等方案，适配不同业务场景：

1. 全局唯一消息ID（雪花算法替代UUID，保证有序不重复）；

2. 基于Redis本地缓存已消费消息ID，设置过期时间，快速拦截重复消息；

3. 核心业务表增加消息ID唯一索引，兜底防重复插入。

import cn.hutool.core.lang.Snowflake;
import cn.hutool.core.util.IdUtil;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 消息幂等工具+修复后消费逻辑
 */
public class MqIdempotentUtil {
    // 本地缓存已消费消息ID（生产可替换为Redis）
    private static final ConcurrentHashMap<String, Long> CONSUMED_MSG_CACHE = new ConcurrentHashMap<>();
    // 消息缓存过期时间 10分钟
    private static final long CACHE_EXPIRE_TIME = 10 * 60 * 1000;
    // 雪花算法生成唯一消息ID
    private static final Snowflake SNOWFLAKE = IdUtil.getSnowflake(1, 1);

    /**
     * 生成全局唯一消息ID
     */
    public static String generateMsgId() {
        return SNOWFLAKE.nextIdStr();
    }

    /**
     * 判断消息是否已消费（幂等校验）
     */
    public static boolean isConsumed(String msgId) {
        // 清理过期缓存
        clearExpireCache();
        return CONSUMED_MSG_CACHE.containsKey(msgId);
    }

    /**
     * 标记消息已消费
     */
    public static void markConsumed(String msgId) {
        CONSUMED_MSG_CACHE.put(msgId, System.currentTimeMillis() + CACHE_EXPIRE_TIME);
    }

    /**
     * 清理过期缓存
     */
    private static void clearExpireCache() {
        long now = System.currentTimeMillis();
        CONSUMED_MSG_CACHE.entrySet().removeIf(entry -> entry.getValue() < now);
    }
}

// 修复后带幂等的消费逻辑
private boolean doConsume(MqMessage message) {
    // 幂等校验：已消费直接返回成功
    if (MqIdempotentUtil.isConsumed(message.getMessageId())) {
        System.out.println("拦截重复消费消息：" + message.getMessageId());
        return true;
    }
    try {
        // 执行业务逻辑
        System.out.println("消费消息：" + message.getMessageId() + "，内容：" + message.getContent());
        TimeUnit.MILLISECONDS.sleep(20);
        // 消费成功标记幂等
        MqIdempotentUtil.markConsumed(message.getMessageId());
        return true;
    } catch (Exception e) {
        return false;
    }
}

坑点5：无死信队列+重试泛滥，导致无效消息永久占用队列

2.5.1 线上故障现象

线上队列长期积压数十万消息，清理后很快再次积压，CPU、内存持续高位。排查发现队列中存在大量永久消费失败的脏消息：参数非法、业务数据不存在、接口永久失效的消息，反复重试、永久失败，无限占用消费线程和队列资源，导致正常消息消费受阻。

2.5.2 问题根因分析

初始重试逻辑漏洞：消息达到最大重试次数后，仅修改状态为死信，未从队列清理、未单独归档、无告警。

这些失效消息会永久驻留内存队列和磁盘文件，每次服务重启都会重新加载，持续占用资源，形成“重试泛滥”问题，慢慢拖垮整个MQ服务。

2.5.3 完整修复方案

1. 新增独立死信队列，超限重试消息转移归档；

2. 死信消息单独持久化，不混入正常消息队列；

3. 死信消息触发告警，人工介入排查；

4. 定时清理过期死信消息，释放资源。

坑点6：线程中断处理不当，引发消费线程死锁、线程泄露

2.6.1 线上故障现象

服务运行1天后，消费线程数持续减少，最终所有消费线程全部挂掉，队列消息彻底停止消费。线程dump分析发现：大量消费线程处于WAITING阻塞状态，无法唤醒，形成线程泄露、局部死锁。

2.6.2 问题根因分析

初始消费代码中，takeMessage()是阻塞方法，线程中断异常处理不规范：捕获中断异常后，仅跳出循环，未重置中断状态，导致线程池无法正常回收线程，最终线程全部卡死。同时JUC阻塞队列的take方法在多消费者场景下，错误的唤醒机制也会导致线程永久阻塞，这是Java并发编程的经典坑点。

2.6.3 完整修复方案

规范中断异常处理，重置线程中断状态，优化线程退出逻辑，避免线程泄露。

坑点7：消息体无版本兼容，迭代升级导致消息解析失败

2.7.1 线上故障现象

服务迭代升级，新增消息扩展字段后，旧版本持久化的历史消息全部解析失败，批量进入死信，导致大量历史数据消费异常。

2.7.2 问题根因分析

初始消息实体无版本号标识，JSON序列化反序列化严格匹配字段，服务升级增减字段后，新旧消息格式不兼容，旧消息解析报错，无法正常消费。这是自研组件迭代最容易忽略的兼容性问题。

2.7.3 完整修复方案

消息实体新增版本字段，序列化开启兼容模式，新旧版本消息平滑适配。

坑点8：无消息超时机制，滞留过期消息持续消费

2.8.1 线上故障现象

队列中存在大量数小时前的过期消息，早已无业务意义，但仍在持续重试消费，浪费大量CPU和IO资源，影响正常业务消息吞吐。

2.8.2 问题根因分析

初始设计无消息过期时间，所有消息永久有效，业务超时、活动结束、订单过期的消息无法自动失效，持续占用队列资源。

2.8.3 完整修复方案

消息实体新增过期时间字段，消费前优先校验消息是否过期，过期消息直接丢弃并归档，不执行消费逻辑。

坑点9：内存队列无内存管控，长时间运行内存泄漏、OOM风险

2.9.1 线上故障现象

服务长时间运行，堆内存持续上涨，无法回收，频繁发生Full GC，严重时触发OOM服务宕机。

2.9.2 问题根因分析

1. 本地缓存无自动清理，幂等缓存、临时数据持续累积；

2. 死信消息、过期消息长期驻留内存；

3. 持久化线程、消费线程无资源回收，句柄泄露；

4. 队列积压消息过多，大量对象常驻堆内存，无法GC。

2.9.3 完整修复方案

新增内存监控、定时清理无效消息、缓存过期机制、资源自动回收逻辑，严控内存占用。

坑点10：无监控告警体系，故障被动发现、事后救火

2.10.1 线上故障现象

所有故障均是业务反馈后才发现，队列积压、消息丢失、消费异常、线程卡死等问题无任何提前预警，故障响应滞后，影响业务稳定性。

2.10.2 问题根因分析

初始自研MQ完全无监控、无日志统计、无告警机制，属于“盲人运行”，无法实时感知队列状态、吞吐、堆积、异常数据。

2.10.3 完整修复方案

新增全方位监控指标：队列积压量、每秒吞吐、消费成功率、重试次数、死信数量、线程状态、内存占用，对接钉钉告警，异常自动触发预警。

三、优化后完整架构与最终落地效果

3.1 优化后完整架构升级

经过线上坑点复盘与迭代优化，最终自研百万级MQ完成全方位架构升级，补齐所有短板：

1. 安全投递层：区分核心/非核心消息投递策略，队列阈值告警，阻塞兜底，杜绝消息静默丢失；

2. 高并发存储层：线程安全批量持久化，文件修复兜底，版本兼容，过期消息自动清理；

3. 稳定消费层：动态扩容线程池，批量消费，规范线程中断处理，杜绝线程死锁和泄露；

4. 可靠性保障层：幂等防重复消费、超时失效、死信归档、失败重试可控；

5. 监控运维层：全指标监控、实时告警、日志溯源、资源自动回收。

3.2 线上最终运行数据

优化上线后，持续稳定运行3个月，无任何故障：

1. 单机峰值吞吐稳定150万+/分钟，较初始提升25%；

2. 消息丢失率、重复消费率、异常率均降至0；

3. 内存、CPU占用稳定，无频繁GC、无内存泄漏；

4. 百万级消息堆积可在1分钟内快速消费完毕；

5. 服务重启、流量峰值、网络抖动等极端场景完全适配。

四、自研消息队列核心血泪经验总结

本次自研百万级MQ的踩坑经历，让我彻底明白：压测完美不代表线上稳定，自研中间件的核心难点从来不是性能，而是边界场景的可靠性。

总结10条可直接落地的自研MQ避坑准则，覆盖所有核心痛点：

1. 永远不要用无锁多线程文件写入，IO操作必须串行化、批量刷盘；

2. 有界队列禁止直接使用offer做核心消息投递，必须做阻塞兜底和告警；

3. 消费线程池必须动态扩容，固定线程池无法适配生产流量波动；

4. 重复消费是必然事件，任何自研MQ必须强制做业务幂等；

5. 必须配置死信队列，无效消息绝对不能留在正常队列中；

6. 线程阻塞、中断异常必须规范处理，否则必然线程泄露死锁；

7. 消息体必须带版本号，预留迭代兼容能力；

8. 所有消息必须配置过期时间，杜绝无效消息占用资源；

9. 必须完善内存、资源管控，长时间运行服务优先防泄漏、防OOM；

10. 无监控的中间件就是裸奔，全指标告警是稳定性的最后防线。

五、自研vs开源MQ 最终选型建议

经过本次实战，给所有开发者最真实的选型建议：

优先选开源MQ的场景：分布式系统、跨服务消息投递、需要高可用集群、事务消息、延时队列、海量消息堆积、无定制化需求的通用业务，直接使用RocketMQ/Kafka，稳定省心，避免重复造轮子。

可以自研MQ的场景：单机服务内异步解耦、轻量化极简需求、需要深度定制逻辑、极致性能优化、无集群高可用诉求、运维资源有限的小型业务。

核心结论：自研MQ的成本不在于开发，而在于线上坑点排查、稳定性打磨、长期迭代维护。如果没有足够的线上故障处理经验，尽量不要自研中间件，看似简单的队列，隐藏的并发、IO、线程、边界问题远超想象。

结尾

这次自研百万级消息队列的踩坑经历，是我Java后端生涯中最深刻的实战复盘之一。看似简单的生产者-消费者模型，在百万级高并发、长时间运行、极端边界场景下，会爆发出无数测试环境无法复现的隐藏问题。

很多时候我们开发的功能，只是实现了“可用”，但距离生产级别的“稳定、可靠、健壮”还有极大的差距。开源中间件的价值，不仅仅是提供功能，更是帮我们屏蔽了无数底层并发、IO、容错、边界的坑。

本文所有代码均为线上修复落地版本，所有坑点均为真实生产故障，希望能帮各位开发者避开自研中间件的误区，少走弯路。