消息队列

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 发过去。消费者同理。 ...

事务消息 本文是分布式算法科普系列第五篇。上一篇讲了 2PC 和 TCC——处理"同步调用"场景下的分布式事务。这一篇换一个跑道——当业务逻辑和消息发送需要原子化，但发消息本身是异步的，怎么保证一致性？ 一、故事：“先写数据库还是先发消息"的终极难题 在消息队列成为微服务通信标配之后，开发者很快撞上了一个死结。一个极其常见的场景：订单创建成功 → 需要发一条消息通知下游（发优惠券、发短信、记录日志）。代码看起来人畜无害： // 伪代码——演示问题——不要在生产里这么写 BEGIN TRANSACTION INSERT INTO orders (...) COMMIT // ↓ 事务已经提交了 mq.send("order_created", order) // 如果这里执行之前——进程突然挂了？ 数据库写入了，消息没发出去——下游永远不知道这笔订单。 那把发消息放进事务里？ BEGIN TRANSACTION INSERT INTO orders (...) mq.send("order_created", order) // 消息队列有自己的事务吗？ COMMIT 数据库事务和消息队列是两套独立的系统——没有"联合事务"这种东西。数据库的 ROLLBACK 不会撤回已经发到 Broker 的消息。 那反过来——先发消息再写数据库？ mq.send("order_created", order) // 消息发出去了 // ↓ 然后写数据库时——数据库挂了 INSERT INTO orders (...) // 失败！ 消息发出去了，数据库没写入——下游收到消息后来查订单——发现根本没有这笔订单。 写过的都懂——这个"先有鸡还是先有蛋"的问题在异步场景下几乎无解。早期方案是在数据库里建一张"消息发件箱"表（outbox），把消息和业务数据在同一个事务里写入，再用一个独立的进程轮询这张表来真正发送。但这个方案太重了——需要额外的轮询进程、需要处理重复投递、需要清理已发送的消息。 2016 年前后，RocketMQ 的团队给出了一个更优雅的方案——让 Broker 自己承担"协调者"的角色，引入"半消息"和"回查"两个机制，一举解决了这个难题。这就是事务消息（Transactional Message）。 二、前置：同步事务 vs 异步事务 在深入事务消息之前，先理清它和上一篇讲的 2PC/TCC 之间的分工： 场景 用哪种方案 特点 服务 A 同步调用服务 B——需要 B 的操作和 A 的操作一起成功或回滚 2PC / TCC 同步——A 等 B 的返回结果 服务 A 发消息给服务 B——需要消息的发送和 A 的本地事务原子化 事务消息 异步——A 不关心 B 什么时候消费 2PC/TCC 处理的是"请求-响应"模式下的分布式事务，事务消息处理的是"发布-订阅"模式下的分布式事务。它们解决的是同一个问题（一致性）的两个不同侧面。 ...

Kafka 生产环境实战 📖 前置阅读：本文是 Kafka 系列的终篇，假设读者已经掌握前五篇的全部内容（核心架构、SpringBoot 集成、Producer 深入、Consumer 深入、Kafka Streams）。 一、⚡ 问题切入：单节点 Docker 的瓶颈 第一篇搭的单节点 KRaft Kafka（一个 Controller + Broker 合体进程）只能用来学习——生产环境中： 单点 后果 唯一的 Broker 挂了 所有消息不可发送/消费——整个 Kafka 瘫痪 无副本 Broker 磁盘损坏 → 消息永久丢失 JVM 内存不足 Full GC 频繁 → 消息延迟抖动 → Producer 超时失败 磁盘写满 Partition 无法写入 → Producer 阻塞或报错 生产最低配：3 台 Broker + 3 台 Controller（或 3 台合体节点，controller + broker 混合模式），每个 Topic 至少 2 副本。 二、KRaft 三节点集群搭建 2.1 架构设计 第一篇用的 KRaft 模式是单节点（Controller 和 Broker 运行在一个进程中）。生产环境拆开： ...

Kafka Streams 流处理实战 📖 前置阅读：本文假设读者已掌握 Kafka Consumer/Producer 的使用和 Offset/Partition 概念。如果还不熟悉，建议先阅读 Consumer 深入：位移管理与 Rebalance。 一、⚡ 问题切入：用 Consumer + Producer 写流处理有什么毛病？ 假设需要统计每个商品的近 5 分钟销量。用 Consumer + Producer 写： // 用 Consumer + Producer 实现滑动窗口计数——代码量爆炸 Map<String, List<Long>> windowCache = new HashMap<>(); // 还需要处理：窗口过期清理、状态持久化、故障恢复、乱序数据... 这个需求正是流处理引擎的用武之地。Kafka Streams 是 Kafka 官方的流处理库——它不是另一个需要部署的服务（不像 Flink/Spark Streaming），而是一个 Java 库，跑在你的应用进程里。 Kafka Streams 本质：Consume → 计算 → Produce，全部走 Kafka，中间状态存在 Kafka 的本地 RocksDB 实例中。 flowchart LR classDef startEnd fill:#701a4c,stroke:#e11d48,stroke-width:2px,color:#fce7f3,font-weight:bold; classDef process fill:#1e1e24,stroke:#6b7280,stroke-width:1.5px,color:#e5e7eb; classDef data fill:#052e16,stroke:#16a34a,stroke-width:1.5px,color:#bbf7d0,font-weight:bold; classDef highlight fill:#450a0a,stroke:#dc2626,stroke-width:1.5px,color:#fecaca,font-weight:bold; INPUT[("input-topic\n原始数据")] -->|"consume"| KS[Kafka Streams\n计算引擎\n+ RocksDB 本地状态] KS -->|"produce"| OUTPUT[("output-topic\n处理结果")] KS -->|"备份"| CHANGELOG[("changelog-topic\n状态变更日志")] class INPUT,OUTPUT data; class CHANGELOG process; class KS highlight; 二、Kafka Streams 核心概念 2.1 KStream vs KTable Kafka Streams 有两个核心抽象，理解它们的区别是正确使用的前提： ...

Kafka Consumer 深入 📖 前置阅读：本文假设读者已掌握 SpringBoot Kafka 的基本消费操作（@KafkaListener）。如果还不熟悉，建议先阅读 SpringBoot Kafka 全操作指南。 一、⚡ 问题切入：消费者重启后，怎么知道上次读到哪了？ RabbitMQ 的答案是"消息消费后就删了，不需要记位置"。RocketMQ 的答案是"Broker 帮你记 offset"。Kafka 的答案是——消费者自己记，记在一个叫 __consumer_offsets 的内部 Topic 里： 消费者在 Partition-2 上消费到 offset=1500 ↓ 提交 offset __consumer_offsets Topic: Key: (order-consumer-group, order-topic, 2) Value: offset=1500 消费者重启 ↓ ↓ 读取 offset 从 offset=1501 继续消费 这个设计是 Kafka 和 RabbitMQ/RocketMQ 最核心的消费端差异——Kafka 的消费者对自己的消费进度负全责。如果消费者忘记提交 offset，重启后就会从上次提交的位置重新消费，产生重复消息。 二、Offset 提交机制 2.1 自动提交 vs 手动提交 Kafka 提供了两种 Offset 提交方式： 提交方式 配置 行为 风险 自动提交 enable-auto-commit: true 每隔 auto.commit.interval.ms（默认 5s）自动提交 poll 返回的最大 offset 消息可能没处理完就提交了——进程挂了会丢消息 手动提交 enable-auto-commit: false + ack-mode: manual 消费者处理完消息后显式调用 ack.acknowledge() 消息可能处理完了但没提交——重启后重复消费 flowchart TD classDef startEnd fill:#701a4c,stroke:#e11d48,stroke-width:2px,color:#fce7f3,font-weight:bold; classDef condition fill:#2a1147,stroke:#a855f7,stroke-width:1.5px,color:#ede9fe,font-weight:bold; classDef process fill:#1e1e24,stroke:#6b7280,stroke-width:1.5px,color:#e5e7eb; classDef highlight fill:#450a0a,stroke:#dc2626,stroke-width:1.5px,color:#fecaca,font-weight:bold; POLL([poll 拉取消息]) --> PROCESS[处理消息] PROCESS --> MODE{提交模式} MODE -- "自动提交" --> AUTO["每隔 auto.commit.interval.ms\n自动提交最后一次 poll 的 offset"] MODE -- "手动提交" --> MANUAL["业务处理成功后\n显式调用 ack.acknowledge()"] AUTO --> RISK1["风险：消息还没处理完\n但 offset 已提交\n→ 进程挂了丢消息"] MANUAL --> RISK2["风险：消息已处理完\n但 offset 没提交\n→ 重启后重复消费"] class POLL startEnd; class MODE condition; class AUTO,MANUAL highlight; class RISK1,RISK2 process; 手动提交比自动提交更安全——至少你知道什么时候提交了。消息重复消费可以用幂等解决，但消息丢失无法恢复。 ...

Kafka Producer 深入 📖 前置阅读：本文假设读者已掌握 SpringBoot Kafka 的基本发送操作（KafkaTemplate.send）。如果还不熟悉，建议先阅读 SpringBoot Kafka 全操作指南。 一、⚡ 问题切入：消息到底发到了哪个 Partition？ 上一篇用 kafkaTemplate.send("order-topic", key, msg) 发送消息时，生产者背后发生了三件事： Partitioner 决定消息进入哪个 Partition 消息攒批——在内存 Buffer 中等待 batch.size 或 linger.ms 条件触发 根据 acks 配置决定什么时候认为发送成功 当你看到日志里 partition=1, offset=0 时，背后是这三个步骤的协作。每个步骤都有配置项可以调整——它们直接影响消息顺序、可靠性、吞吐量。 二、分区策略 —— 消息路由的第一个环节 2.1 默认分区策略 Kafka Producer 的 Partitioner 接口决定了每条消息进入哪个 Partition。默认实现是 DefaultPartitioner： // DefaultPartitioner 的逻辑（简化版） public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) { int numPartitions = cluster.partitionsForTopic(topic).size(); if (keyBytes == null) { // Key 为 null → 使用 Sticky 分区（粘性分区） // 不是轮询！是把一批消息都发到同一个 Partition，等 batch 满了才换下一个 return stickyPartition(topic, numPartitions); } else { // Key 不为 null → 用 murmur2 哈希 % Partition 数量 return Utils.murmur2(keyBytes) % numPartitions; } } 规则一：Key 为 null → Sticky Partition。Kafka 2.4 之前是轮询（Round Robin——每条消息换一个 Partition），2.4+ 改为 Sticky——把一批消息"粘"在同一个 Partition 上，等这个 Batch 满了或时间到了才切到下一个 Partition。这减少了网络请求次数——把同一批的消息打成一个请求发给同一个 Broker。 ...

Kafka：分布式提交日志，不是消息队列 📖 前置阅读：本文假设读者已理解消息队列的基本价值（异步、解耦、削峰填谷），最好读过 RabbitMQ 或 RocketMQ 的任意一篇基础文章。有了 MQ 概念再学 Kafka 事半功倍。 一、⚡ 问题切入：RabbitMQ 和 RocketMQ 有什么共同的"毛病"？ 先回顾 RabbitMQ 和 RocketMQ 的消费模型： RabbitMQ: Consumer 收到消息 → 手动 basicAck → Broker 删除消息 RocketMQ: Consumer 收到消息 → 返回 CONSUME_SUCCESS → offset 推进 共同点：消息被消费者确认（ACK）后，Broker 就把它删了。消息在 Broker 上的生命周期是"暂存"——它存在只是为了等待消费者拿走。 这个模型有一个隐含的限制：一条消息只能被消费一次。想重放消息？RabbitMQ 做不到（消息已经删了），RocketMQ 可以重置 offset 但受 CommitLog 保留时间限制。 这时候再看 Kafka 的设计：消息消费后不删除。消息存在磁盘上，按时间或大小策略统一过期，消费者想从哪个位置读就从哪个位置读。 Kafka: Consumer 自己管 offset，随时可以回到过去的某个位置重读 消息不是被消费掉的——是按时间自然过期的 这就是 Kafka 和 RabbitMQ/RocketMQ 本质上的不同——Kafka 不是一个消息队列，它是一个分布式提交日志（Distributed Commit Log）。 二、🧬 Kafka 是什么：分布式提交日志 2.1 核心定义 Kafka 的官方定位：分布式、分区化、多副本的提交日志服务。 ...

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： ...