RocketMQ

RocketMQ 避坑全攻略 0. 前言 0.1 为什么写这篇博客 消息中间件是分布式系统的必修课——这话没错，但很多团队引入 MQ 的时候只看到了"解耦"和"削峰填谷"的好处，却没意识到它同时带来了消息丢失、重复消费、积压、顺序错乱等一系列新问题。坦白说，踩过这些坑的开发者不在少数。 某开发者在生产环境第一次遇到 RocketMQ 积压十几万条消息的时候，第一反应是重启消费者——结果毫无悬念地失败了。后来花了一整天排查，问题竟然只是消费逻辑里多了一个 Thread.sleep(200) 。这种教训值得记下来。 0.2 读者需要的基础知识 用过 RocketMQ（至少本地跑过 Demo，知道 Producer / Consumer / Topic / Broker 是什么） 知道什么是生产者、消费者、Topic、Broker、NameServer 了解基本的分布式系统概念（如 CAP、最终一致性） 0.3 文章结构说明 按"问题 → 原因 → 原理 → 解决方案"的结构展开，每个问题独立成章。读者可以按需跳读，也可以从头串下来形成体系。 0.4 一句话总结 本文不是教"怎么用"，而是教"怎么用好、怎么避坑"。 默认配置在生产环境就是定时炸弹。 1. 消息丢失 消息丢失是 MQ 使用中最致命的问题之一——订单丢了就是资损，通知丢了就是客诉。先按链路拆解一下丢消息的三个位置。 1.1 丢失场景分类 flowchart TD start([消息发送]) --> prod{生产端是否可靠？} prod -->|网络超时| prodLoss[生产端丢失] prod -->|异步未回调| prodLoss prod -->|重试耗尽| prodLoss prod -->|发送成功| broker[Broker 存储] broker --> persist{持久化策略？} persist -->|异步刷盘| brokerLoss[Broker 端丢失] persist -->|异步复制| brokerLoss persist -->|磁盘故障| brokerLoss persist -->|同步落盘| consume[消费端拉取] consume --> ack{ACK 策略？} ack -->|自动提交| consLoss[消费端丢失] ack -->|并发异常| consLoss ack -->|手动确认成功| done([消息可靠送达]) classDef startEnd fill:#701a4c,stroke:#e11d48,stroke-width:2.5px,color:#fce7f3,font-weight:bold; classDef condition fill:#2a1147,stroke:#a855f7,stroke-width:2px,color:#ede9fe,font-weight:bold; classDef process fill:#1e1e24,stroke:#6b7280,stroke-width:2px,color:#e5e7eb; classDef reject fill:#450a0a,stroke:#dc2626,stroke-width:2px,color:#fecaca,font-weight:bold; classDef data fill:#052e16,stroke:#16a34a,stroke-width:2px,color:#bbf7d0,font-weight:bold; class start,done startEnd; class prod,persist,ack condition; class prodLoss,brokerLoss,consLoss reject; class broker,consume process; 1.1.1 生产端丢失 场景 原因 网络超时 客户端以为发送成功，实际 Broker 未收到。网络抖动 + 超时配置不合理导致 异步发送未回调 producer.send(msg) 后直接 return，异常被吞掉 重试耗尽 RetryTimesWhenSendFailed 次数用完，消息被丢弃 1.1.2 Broker 端丢失 场景 原因 异步刷盘 消息写入 PageCache 后返回成功，但尚未落盘，断电即丢 主从异步复制 Master 宕机时 Slave 未同步到最新数据 磁盘故障 物理损坏导致已落盘数据不可恢复 1.1.3 消费端丢失 场景 原因 自动提交 Offset consumeMessageBatchMaxSize 拉了一批消息，自动 ACK 后业务处理失败 并发消费异常 多线程中某条消息处理失败，但整体无法回滚 1.2 原理深度剖析 1.2.1 存储架构 RocketMQ 的存储核心是 CommitLog（顺序写文件）+ ConsumeQueue（按 Topic+Queue 建索引）+ IndexFile（按 Key 查询）。所有消息先追加到 CommitLog，再异步构建 ConsumeQueue 索引。 ...

RocketMQ 存储模型拆解：别再拿它当 BlockingQueue 用了 0. 引言：一个 Javaer 的认知崩塌 每一个刚接触 RocketMQ 的 Java 开发者，大概都会经历一次认知崩塌： 翻开 RocketMQ 源码，找遍所有 package，也找不到一个像 BlockingQueue 那样的 Queue 实现。作为一个"消息队列"（Message Queue），它的 Queue 到底在哪里？如果找不到 Queue 对象，消息是怎么"入队"和"出队"的？ 答案很残酷，但也很优雅：RocketMQ 根本没有什么 Queue 数据结构。 RocketMQ 的 Queue 是一个文件夹——硬盘上的文件夹。消息不是"入队"，而是 append 到磁盘文件末尾。消费者不是"出队"，而是从磁盘文件读取一段定长的字节数组。 整个 RocketMQ，本质上就是一套精心设计的磁盘文件操作方案。 所有的分布式消息特性——消息重试、死信队列、消息积压、限流熔断——都是在这套磁盘文件模型上玩出的花样。理解了 RocketMQ 的文件长什么样、文件之间怎么关联，你就理解了 RocketMQ 的一切。 这篇从最底层开始，一层层往上拆，共七层： NameServer（路由层）——只管路标，不存消息 CommitLog（存储层）——所有消息顺序写入的大文件 ConsumeQueue（索引层）——被误称为队列的定长指针数组 Topic 参数（运维配置）——目录数量和权限的配置映射 高级特性——基于文件模型的策略实现 开发视角——你的代码在操作什么 总结——一张物理模型图覆盖所有概念 1. 第一层：NameServer（路由层） NameServer 是 RocketMQ 的路由层。它是一个轻量级的注册中心——只维护 Topic 到 Broker 地址的映射关系，不存储任何消息数据。 传统模式 在 RocketMQ 的传统模式下，NameServer 内存中维护着两张核心映射表： BrokerData：Topic 名称 → Broker IP 列表。消费者拿到这个列表，就知道该从哪台机器拉数据。 QueueData：Queue 数量 + 权限设置。Broker 上报时携带每个 Topic 配置了多少 Queue。 生产者发消息时先问 NameServer：“TopicA 在哪台机器上？有几个 Queue？” NameServer 返回 Broker 地址列表，生产者选择一个 Broker 发过去。消费者同理。 ...

RocketMQ 生产部署 📖 前置阅读：本文是 RocketMQ 系列的终篇，假设读者已经掌握前五篇的全部内容（核心架构、SpringBoot 集成、高级消息类型、可靠性、消费者模式）。 一、⚡ 问题切入：单机 Docker 的瓶颈 第一篇搭的单机 RocketMQ（一个 NameServer + 一个 Broker）只能用来学习——生产环境中： 单点 后果 一台 NameServer 挂了 Producer/Consumer 无法获取路由——整个 MQ 瘫痪 一台 Broker 挂了 所有消息不可用——消息无法发送/消费 JVM 内存不足 Full GC 频繁 → 消息延迟抖动 → 超时重试雪崩 磁盘写满 CommitLog 无法写入 → 生产者阻塞 生产最低配：2 台 NameServer + 至少 2 台 Broker（主从）。 二、双主双从高可用集群搭建 2.1 架构设计 NameServer 集群：2 台（互不通信，各自独立） Broker 集群：2 组主从（Master-A + Slave-A、Master-B + Slave-B） NameServer-1 NameServer-2 (192.168.1.10:9876) (192.168.1.11:9876) ↑ ↑ ┌─────────┴───────┬───────────────┘ │ │ Broker-A (Master) Broker-A (Slave) 192.168.1.20:10911 192.168.1.21:10911 brokerId=0 brokerId=1 Broker-B (Master) Broker-B (Slave) 192.168.1.22:10911 192.168.1.23:10911 brokerId=0 brokerId=1 路由发现：Producer/Consumer 配置所有的 NameServer 地址——只要有一台 NameServer 活着，路由就能工作。 ...

RocketMQ 消费者模式 📖 前置阅读：本文假设读者已掌握 SpringBoot RocketMQ 的基本消费操作。如果还不熟悉，建议先阅读 SpringBoot RocketMQ 全操作指南。 一、⚡ 问题切入：一条消息，谁来消费？ 前面五篇的消费者代码都默认了一件事——一条消息只被一个消费者实例处理。但实际业务中： 订单消息——只能被一个实例消费（一个订单不能被两个服务处理两次） 配置刷新消息——所有实例都要收到（所有缓存节点刷新缓存） 部分 Tag 的消息——只关心"订单创建"，对"订单支付"不感兴趣 消费不过来——10 个 Queue，2 个实例，怎么分？ 这四大问题的答案都在这一篇里。 二、集群消费 vs 广播消费 2.1 集群消费（CLUSTERING）——默认模式 同一个 ConsumerGroup 内的所有实例共享消费一个 Topic 的消息——每条消息只被组内一个实例消费。 @Component @RocketMQMessageListener( topic = "order-topic", consumerGroup = "order-consumer-group", messageModel = MessageModel.CLUSTERING // 集群模式（默认值，可以不写） ) public class OrderClusteringListener implements RocketMQListener<OrderMessage> { @Override public void onMessage(OrderMessage msg) { System.out.printf("[实例A] 处理订单: orderId=%d%n", msg.getOrderId()); } } 部署 3 个实例，Topic 有 8 个 Queue： ...

RocketMQ 消息可靠性 📖 前置阅读：本文假设读者已掌握 SpringBoot RocketMQ 的发送和消费操作。如果还不熟悉，建议先阅读 SpringBoot RocketMQ 全操作指南。 一、⚡ 消息可能丢在哪？ 和 RabbitMQ 一样，RocketMQ 的消息丢失也分三个环节： Producer → [网络] → Broker → [网络] → Consumer ① 发送丢失 ② Broker 宕机丢失 ③ 消费失败丢失 在 RocketMQ 中，生产者端没有 RabbitMQ 的 Publisher Confirm——替代方案是同步发送 + 返回值判断。Broker 端的持久化取决于刷盘策略和主从同步。消费端靠消费状态返回 + 重试 + 死信。 二、① 生产者端：同步发送判断返回值 RocketMQ 没有 RabbitMQ 的 ConfirmCallback 异步通知机制，但 syncSend 本身就是同步等待 Broker 确认——返回 SEND_OK 说明消息已写入 CommitLog： @Service public class ReliableProducer { @Autowired private RocketMQTemplate rocketMQTemplate; public void sendReliably(OrderMessage msg) { // syncSend 是同步阻塞的——返回 SEND_OK 才说明 Broker 已接收 SendResult result = rocketMQTemplate.syncSend( "order-topic:created", msg); if (SendStatus.SEND_OK.equals(result.getSendStatus())) { log.info("消息已确认到达 Broker: msgId={}", result.getMsgId()); } else { // 只有 SEND_OK 才认为成功——FLUSH_DISK_TIMEOUT 等状态说明刷盘超时 log.error("消息发送未确认: status={}", result.getSendStatus()); // 补偿：写入 DB 重试表 saveToRetryTable(msg); } } } SendStatus 的四种返回值： ...

顺序消息、延迟消息与事务消息 📖 前置阅读：本文假设读者已掌握 SpringBoot RocketMQ 的基本操作（RocketMQTemplate、@RocketMQMessageListener）。如果还不熟悉，建议先阅读 SpringBoot RocketMQ 全操作指南。 一、⚡ 问题切入：三种 RabbitMQ 做不到或做不好的事 RabbitMQ 六篇系列学完时留了几个坑——有些场景 RabbitMQ 不是不能用，而是做起来别扭： 需求 RabbitMQ 方案 痛点 订单创建→支付→发货严格按序 单队列 + 单消费者，关并发 吞吐量压到一条线；一旦重试入队顺序全乱 30 分钟后自动取消 Delayed Message 插件或 TTL+DLX 插件生产不可靠；TTL+DLX 有消息时序问题 下单 + 扣库存 + 发消息三件事原子执行 自己实现本地消息表 + 定时补偿 代码量大，维护麻烦 RocketMQ 对这三种场景都有原生支持——不是插件，不是 workaround，是设计时就考虑进去了。 二、顺序消息 —— 深度篇 2.1 上一篇回顾 + 补充 上一篇讲了基本用法：syncSendOrderly 用 orderId 哈希选 Queue，同一个 orderId 进同一个 Queue → 该 Queue 内 FIFO。消费端 consumeMode = ConsumeMode.ORDERLY。 但这只讲了正常流程。重试会破坏顺序——这是最容易踩的坑。 2.2 顺序消费的重试机制：挂起而非重入队 并发消费中，失败的消息通过 RECONSUME_LATER 进入重试 Topic，然后延迟重新投递。但顺序消费不能这么干——如果第 2 条消息失败后进了重试队列，第 3 条消息先被消费，顺序就乱了。 ...

SpringBoot 集成 RocketMQ：从发送到消费 📖 前置阅读：本文假设读者已理解 RocketMQ 的核心概念（NameServer、Broker、Topic、Queue、ConsumerGroup）。如果还不熟悉，建议先阅读 RocketMQ 核心架构与消息模型。 Part 1：概念与前置 1.1 本文目标 上一篇用原版 RocketMQ Java Client 写了 DefaultMQProducer + DefaultMQPushConsumer。真实的 SpringBoot 项目里不需要那么多样板代码——rocketmq-spring-boot-starter 帮你处理了 NameServer 连接、Producer 启动、Consumer 注册。 读完这篇会掌握： MqHelper 封装——为什么要在 RocketMQTemplate 上再包一层，asyncSend + SendCallback 的真实用法 30 分钟延迟取消订单——RocketMQ 内置 delayLevel 的完整实战流程，和 RabbitMQ 方案的对比 @RocketMQMessageListener 消费者——MessageExt 手动反序列化 vs 泛型自动解析，以及三个真实业务消费者 Domain Entity 直传——为什么不做 DTO 转换，以及什么情况下不能这样做 双 MQ 基础设施先行——RabbitMQ 拓扑已就绪但全用 RocketMQ 的设计决策 1.2 前置条件 前置项 具体要求 验证命令 JDK 17+（8+ 也兼容） java -version SpringBoot 3.x（文中用 3.2） mvn dependency:tree | grep spring-boot RocketMQ 5.1.4（NameServer + Broker 都在运行） docker ps | grep rocketmq 前置知识 NameServer/Broker/Topic/Queue 概念 — 确认 RocketMQ 在跑： ...

领域模型与存储引擎 📖 前置阅读：本文假设读者已理解消息队列的基本价值（异步、解耦、削峰填谷）。如果还不熟悉消息队列，建议先阅读 RabbitMQ 核心概念与 AMQP 协议。 一、问题切入：为什么 RocketMQ 的概念比 RabbitMQ 多？ RabbitMQ 学完六篇，Exchange / Binding / Queue 的路由模型印象深刻——概念不多，全靠灵活组合。翻开 RocketMQ 的文档，一眼扫过去：Producer、Consumer、Topic、Queue、ConsumerGroup、Subscription、Broker、NameServer……光领域概念就七个，外加两个部署组件。 这不是设计过度，而是 RocketMQ 把"谁负责发、谁负责收、怎么分组、怎么扩容、消息存哪里、谁管路由"全部显式拆开了。 RabbitMQ 用少数概念的组合来表达这些维度，RocketMQ 选择每件事都定义一个独立概念。 好处是每个概念职责单一，坏处是初学者一看就晕——概念之间谁包谁、谁管谁、谁和谁是平等的，不看图根本理不清。 所以这篇不讲"先记住七个概念"——先看一张全景图，把七者的层级关系钉在脑子里。 二、领域模型全景：一张图串起七个概念 Apache RocketMQ 官方把领域模型定义为七个核心概念。这是一张按层级包含关系组织的全景图： flowchart TD subgraph PRODUCTION["① 生产"] P(["Producer\n生产者"]) end subgraph STORAGE["② 存储"] MSG["Message\n消息体"] TOPIC["Topic\n主题（逻辑容器）"] Q0[("Queue-0\n物理分片")] Q1[("Queue-1")] Q2[("Queue-n")] MSG --> TOPIC TOPIC --> Q0 TOPIC --> Q1 TOPIC --> Q2 end subgraph CONSUMPTION["③ 消费"] CG["ConsumerGroup\n消费分组"] C1(["Consumer-1"]) C2(["Consumer-2"]) SUB["Subscription\n订阅关系"] CG --> C1 CG --> C2 CG --> SUB end P -->|"发送"| TOPIC Q0 & Q1 & Q2 -->|"拉取"| CG SUB -.->|"绑定"| TOPIC classDef startEnd fill:#fff5f5,stroke:#e53e3e,stroke-width:2.5px,font-weight:bold; classDef process fill:#f7fafc,stroke:#718096,stroke-width:2px; classDef data fill:#f0fff4,stroke:#38a169,stroke-width:2px,font-weight:bold; classDef root fill:#ebf8ff,stroke:#3182ce,stroke-width:2.5px,font-weight:bold; class P,C1,C2 startEnd; class MSG,TOPIC process; class Q0,Q1,Q2 data; class CG,SUB root; 这张图的阅读顺序：从左到右，消息从 Producer 出发 → 经过 Topic → 落入 Queue → 被 ConsumerGroup 内的 Consumer 拉取。关键层级关系： ...