rocketmq 重复拉取消息_RocketMQ 分享全纪实

朴灿烈づ我的快乐病毒、 2023-01-03 01:27 113阅读 0赞

![ca57a04c9932ba80763952c4aa316fa1.png][]

> 这是公众号 Young\_Blog 的第 41 篇文章  
> 封面图片来自 Jezael Melgoza

*  为什么分享 RocketMQ
 *  消息队列的组成部分
 *  RocketMQ 的架构
    
     *  Name Server
     *  生产组和消费者
     *  Topic 和 MessageQueue
     *  Broker
 *  RocketMQ Producer
    
     *  消息发送的负载均衡
     *  消息发送的高可用
 *  RocketMQ Consumer
    
     *  Message Queue 的分配
     *  保证局部顺序消费
     *  管理消费进度
     *  Rebalance
 *  RocketMQ 消息存储
 *  总结

### 为什么分享 RocketMQ ###

为什么没有选择 `Kafka` 而是 `RocketMQ` 呢，没有什么特别的原因，单纯是我之前就看过一点 `RocketMQ` 的源码，但是后来因为各种原因没能看完，因此想着趁这次机会系统地回顾一遍。

另外就是，之前我的工作集中在客户端或者服务端，很少端到端地去设计、开发某项功能，因此在架构设计方面积累的经验比较少。这点劣势在年初接手开发新项目的时候尤为明显，新项目涉及注册发现、客户端、服务端、监控、网络编程等一系列模块，因为之前没有接触过这么复杂的项目，也不了解业务成熟的大型项目的设计哲学，因此在轮到自己做架构设计的时候提出的方案总是存在一些缺陷，后续导致了返工的现象。

那时起我就琢磨着找一个大型开源项目研究一下，学习一下它在架构设计上的考量，`RocketMQ` 不论从规模还是复杂度上讲都很合适，因此就选了它。

> 最近腾讯开源了 TubeMQ，感兴趣的可以去看看 TubeMQ

### 消息队列的组成部分 ###

首先，明确下消息队列的设计目标，通常有以下几点：

1.  解耦，需要支持消息暂存
2.  削峰，要求能容忍消息积压
3.  保证消息尽可能不丢、不重复消费
4.  支持传统意义上的队列和广播两种消费模式
5.  消费能力得可以横向扩展

貌似要求越写越多，那设想一下，如果从零开始设计一个消息队列，满足以上五个需求，需要哪些组成部分。

第一个很容易想到的就是生产者（`Producer`）和消费者（`Consumer`），这是消息队列两端最重要的组成部分。通常情况下，生产者和消费者都不会是单台机器，而是一个集群，因此用生产组（`Producer Group`）和消费组（`Consumer Group`）来描述。

紧接着为了允许消息暂存和消息积压，我们通常需要一个消息代理服务器（`Broker`），生产者和消费者都仅连接 `Broker` 进行消息的发送和消费，二者之间完全解耦。

消息发送通常需要一个管道进行投递，而且不同类型的消息最好发往不同的管道，管道的另一端连接消费者，通过增加管道数目和消费者数目，我们就达到了横向扩展消费能力的目的，这里我们将管道成为 `Queue`，将消息类型成为 `Topic`。

很好，目前我们搭建了 **生产者 -> Queue -> Broker -> Queue -> 消费者** 的完整流程，一条消息已经可以端到端地发送了。但是思考下，如果 `Topic` 和 `Queue` 的数目很多，某些生产者和消费者只关注其中一些，那么我们还需要为这种订阅/发布关系提供一个注册平台，称之为 `NameService`（命名服务）来统一管理消息订阅的拓扑关系。

上面废了这么多口舌，**目的就在于让不了解消息队列的同学在进入下一节之前，不至于一脸懵逼**。技术的架构设计都是一步一步来的，消息队列也不是一开始就演变成 `Kafka` 或者 `RocketMQ` 这种架构的，它们都经历过为了支持某些业务需求而不得不做的架构演进，最后变成了现在目前业界比较成熟的模型，而我们上述的假设其实已经走了很多捷径。

### RocketMQ 的架构 ###

这节我们来看看 `RocketMQ` 的具体架构，结合这张结构图我们会介绍里面每个组成部分的功能和设计考量。

![3218bcba81ddd2644643ac3a88948563.png][]

RocketMQ 架构图

### Name Server ###

位于最顶端的是 `RocketMQ` 的命名服务，称之为 `NameServer`，它是用来管理 `Topic` 的订阅发布关系、消息发送和消费拓扑的，得让生产者知道 “我这个 `Topic` 的信息发往哪些 `Broker`”，得让消费者知道 “我得去哪些 `Broker` 上消费这个 `Topic` 的消息”。`NameServer` 可以多机部署变成一个 `NameServer` 集群保证高可用，但这些机器间彼此并不通信，也就是说三者的元数据舍弃了强一致性。

这些元数据是怎么来的呢，首先 `Broker` 启动时会向全部的 `NameServer` 机器注册心跳，心跳里包含自己机器上 `Topic` 的拓扑信息，之后每隔 `30s` 更新一次，然后生产者和消费者启动的时候任选一台 `NameServer` 机器拉取所需的 `Topic` 的路由信息缓存在本地内存中，之后每隔 `30s` 定时从远端拉取更新本地缓存。

`NameServer` 机器中定时扫描 `Broker` 的心跳，一旦失联超出 `2min`，即关闭这个 `Broker` 的连接，但不主动通知生产组和消费组，因此二者最长需要 `30s` 才能感知到某个 `Broker` 故障。

### 生产组和消费者 ###

架构图两端的就是生产组和消费组，都是多机的集群，由若干个生产者和消费者实例组成，消费者消费消息时有两种模式，一种是广播模式一种是集群消费模式，前者表示一条消息会被消费组下的所有消费者实例消费，后者表示一条消息只会被消费组下的一个实例消费到，考虑到集群消费模式是目前使用主流，因此本文主要谈论后者。

### Topic 和 MessageQueue ###

`Topic` 我们之前讲过了，代表某种消息类型，为了达到消费性能可横向扩展的需求，`RocketMQ` 引入了 `MessageQueue` 这个逻辑概念，将一个 `Topic` 划分为多个 `MessageQueue`，默认是四个。而 `MessageQueue` 和消费者实例是一对一的关系，消费者实例和 `MessageQueue` 是一对多的关系。

例如架构图中，`Topic` 下分为四个 `MessageQueue`，分布在两个 `Broker` 机器上，生产者组将消息平均发往四个 `MessageQueue`，而由于消费组中仅有两个消费者实例，因此每个消费者实例平均消费两个 `MessageQueue`。

一旦性能不足，可以扩容消费组增加消费者实例至四个，那么每个消费者实例消费一个 `MessageQueue`，从而达到消费能力的横向扩展。

### Broker ###

`Broker` 作为消息代理服务器，最重要的职责是存储消息和管理消费进度（集群消费模式下专有）。单个 `Topic` 下的多个 `MessageQueue` 一般来说会分散在多个 `Broker` 上面达到容灾的目的。

`Topic` 通过打散 `MessageQueue` 达到容灾目的，那么 `Broker` 机器维度又是怎么容灾的呢，`RocketMQ` 允许设置主备 `Broker`，二者间通过异步拉取复制的方式进行消息同步，一旦主 `Broker` 宕机，备机可以提供消息消费，但不提供消息写入，也就是说其实主备之间并没有 `Failover` 功能，这保证了写入主的消息不会丢失，但是会影响系统的可用性。

滴滴内部做过针对性地做过二次开发，简单来说实现的方式是 `NameServer` 集群通过 `ZK` 选举出一个 `Leader`，来完成 `Failover` 的决策。

> 为了简洁，本文图例中没有说明的情况下，均不画出 Slave Broker

### RocketMQ Producer ###

### 消息发送的负载均衡 ###

![3218bcba81ddd2644643ac3a88948563.png][]

还是看这张结构图，生产者发送消息，默认采用轮询的方式达到负载均衡，每个生产者实例内存中都知道 `Topic` 下 `MessageQueue` 的分布拓扑信息，因此通过轮询就可以将消息平均发送到这些管道里。

### 消息发送的高可用 ###

我们之前提到过，`Broker` 会向 `NameServer` 集群所有机器发送心跳，`NameServer` 集群里的机器各自定期扫描失联的 `Broker`，关闭连接，但不会主动通知生产者组，需要等待生产者主动来拉取。因此存在元数据不一致的窗口，此窗口最长为 `30s`。

![928cbb8f762af35740161b3c2fa54357.png][]

由于上述原因，消息生产者不可避免的会将消息发往已经故障的 `Broker` 机器，例如上图，`Producer-01` 先将消息发往 `Broker-A` 上的 `MessageQueue-01`，发现失败了，由于轮询发送机制它继续发往 `MessageQueue-02`，由于还是位于 `Broker-A` 机器上，因此依旧失败了，默认情况下同步发送消息重试三次，因此很可能这条消息由于没有规避 `Broker-A` 导致发送失败，实际上 `Broker-B` 还是存活的，完全可以规避掉故障的 `Broker-A` 机器提前选择 `Broker-B` 发送消息。

![1df14cdc8b830b8df33dab08abd641a1.png][]

`RocketMQ` 中将生产者端剔除故障机器的机制称之为 `Broker` 的故障延迟机制，一旦发现发送到某个 `Broker` 机器失败，则暂时将其剔除，优先选择其他 `Broker` 重试。

### RocketMQ Consumer ###

看完了消息发送部分，本节我们进入消息的消费。消息的消费相较于消息的发送会复杂一些。我们想一下，假设你某个生产者实例宕机了，那最多就是少了个消息的发送者，而绝大多数情况下消息的生产者都是无状态的，流量可以任意打到某个生产者，如果其一宕机那么我通过一些措施摘掉这台机器的流量就可以。但是消费者没有这么简单，因为它们并不是无状态的，它们是固定在消费某一些 `Topic` 的 `MessageQueue`，因此宕机任意一台消费者都涉及到消费拓扑的重新变更，这带来了更多的复杂度。

### Message Queue 的分配 ###

`MessageQueue` 存在的意义前面已经谈过不再复述，本节讲一下如果将特定数量的 `MessageQueue` 分配给消费者组下的消费者实例，注意！这其实是个技术活。

消费者组下的消费者实例，怎么知道自己需要消费某个 `Topic` 下的哪些 `MessageQueue` 呢？例如架构图中，只有两个消费者实例，但是总共有四个 `MessageQueue`，他们如何知道各自消费两个，而且还没有冲突的。

> 为了简单，假设我们的系统是新搭建的，两台 Consumer 都是第一次启动，因此这里不涉及 Rebalance 机制

分配方案是在消费者实例启动的时候去执行的，消费者实例启动的时候回从 `NameServer` 上获取自己订阅的 `Topic` 的拓扑信息，包括该 Topic 下总共有几个 `MessageQueue`，分布在哪些 `Broker` 机器上等等。然后向其中所有 `Broker` 机器发送心跳。最后选取任意一台 `Broker`，从上面获取消费组下总共有几个实例。

![74f6e1eec17649407716d3b827ae557a.png][]

如此一来，消费者实例就知道了 `MessageQueue` 信息（`mqSet`）和消费组下的实例个数（`consumerIdSet`）信息。在本地内存中通过简单的分配算法，就可以知道自己应该负责消费哪些 `MessageQueue` 了。

需要注意的是，每个客户端获取到 `mqSet` 和 `consumerIdSet` 之后都需要首先进行排序！目的是为了在执行分配算法时，每个客户端的视图都是一致的。

![dba9484946266f630cdbfed896f79bf7.png][]

`RocketMQ` 针对 `MessageQueue` 提供了多种可选的分配策略，例如平均分配、轮询分配、固定分配等，在实际生产环境中可能还需要根据机房进行就近路由分配、粘滞分配（使得 `MessageQueue` 变动次数最小）等。

### 保证局部顺序消费 ###

顺序消费是应用场景对消息队列中间件提出的需求，例如某个 `ID = 100` 的支付业务，在其生命周期内会发送三条消息：

1.  订单生成
2.  订单付款
3.  订单结束

因为订单 `ID` 同为 `100` 属于一个订单，因此要求消费组在消费这三条消息时保证先消费第一条，然后才能消费第二条，最后才是第三条。如果此时还有 `ID = 300` 的订单，那么二者之间可以交叉，但是这三个过程必须保证升序。

![40c798062228f30111daac2f6dd2b43b.png][]

保证消息局部顺序消费的重点在于：

1.  生产者组通过计算，将相同 `ID` 的订单消息发往同一个 `MessageQueue`
2.  消费者组通过分别位于 `Broker` 和客户端的两把锁，保证对该 `MessageQueue` 内消息的顺序消费

发往同一个 `MessageQueue` 保证了该 `MessageQueue` 内消息是局部有序的，但是无法保证全局有序，想要全局有序？那这个 `Topic` 只能配一个 `MessageQueue`，然后全部消息都发到这一个 `MessageQueue` 中。一般来说，局部有序已经可以满足绝大部分应用场景了。

生产端的保证达到了，下面就是消费端，依靠的是两把锁，分别位于 `Broker` 侧和消费者实例客户端侧。`Broker` 侧的锁是 `MessageQueue` 粒度的，保证同一时间至多只有一个消费者实例消费该 `MessageQueue`。

![78f6345ed895b9cfa08570a79f117b00.png][]

你可能疑惑，本来不就是一对一的关系么？原因是在消费者组进行 `Rebalance` 的时候可能会造成某个时间窗口内单个 `MessageQueue` 被多个消费者实例同时消费，这里通过加锁限制了这种情况。一旦启动时加锁失败，意味着该 `MessageQueue` 还在被其他消费者实例锁定，因此不创建相应的消息拉取任务，等到锁被释放或者超时（默认 `60s`）。加锁成功后消费者实例还会每隔 20s 定时锁定该 `MessageQueue` 一次。

消费者实例侧由于可能同时负责消费多个 `MessageQueue`，因此采用了线程池消费消息，需要在客户端提供加锁的方式保证单个 `MessageQueue` 内的消息同一时间仅被一个线程消费。

### 管理消费进度 ###

在广播消费模式下，消费进度仅存储在消费者实例本地，而在集群消费模式下，消费进度存储在 `Broker` 上。通过 `Topic + 消费者组名称`作为 `key`，`value` 中分别记录每个 `MessageQueue` 对应该消费者组的消费偏移量，因此消费进度是消费者组之间互相隔离的。

> 早期版本 `Kafka` 将 `offset` 保存在 `ZK` 上， `Path` 为 consumers/\{consume-group\}/offsets/\{topic\}/\{partition\}，其实和 `RocketMQ` 的保存方式是一致的

利用 `offset` 记录消费进度本质上是一种批量 `ACK` 的方法，它的优点在于 `Broker` 的消费进度管理粒度从单条消息变为单个 `MessageQueue`，简化了 `Broker` 的复杂度。那么下一个问题，消费者和 `Broker` 都是在何时提交和持久化各自的 offset 的呢？

首先，消费者侧会记录自己的消费进度到内存中的 `OffsetTable`，通过每五秒一次的定时任务提交到 `Broker` 侧，`Broker` 接收到之后保存在内存中，并定时刷到磁盘上的 `json` 文件里。

这里需要注意的是，由于一批消息的消费次序不确定，可能下标大的消息先被消费结束，下标小的由于延时尚未被消费，此时消费者向 `Broker` 提交的 `offset` 应该是已被消费的最小下标，从而保证消息不被遗漏，但缺点在于可能重复消费消息。

![ddcab2bae82c0f796d80371faeb3977f.png][]

### Rebalance ###

消息队列系统中，经常会出现 `Broker` 实例的增删、`Topic` 的增减、`Topic` 下 `MessageQueue` 数目的增减、消费组实例数目的增减等情况，它们都会触发消费关系的重新分配，这个过程称之为 `Rebalance`。

`RocketMQ` 的 `Rebalance` 机制有主动和被动之分，主动意为消费者实例每隔 `20s` 会定时计算自己的消费拓扑并和内存中的对比，一旦发现部分 `MessageQueue` 不再是自己负责消费，则停止对它的消息拉取任务；如果有新的 `MessageQueue` 变为自己负责，则创建对它的消息拉取任务。

被动意为，`Broker` 可以主动通知某个消费组下的所有实例，要求它们立即开始一次 `Rebalance`，常用于新的消费者实例加入、或者 Broker 检测到有消费者实例心跳失联等情况，下面是一个消费者实例新加入的场景。

![8ca0c7f4ff9e2807cb72f1f4d5a2324f.png][]

`RocketMQ` 的 `Rebalance` 由于部分时刻的视图可能存在不一致，因此单次 `Rebalance` 并不能完全保证一定达到最终效果，但是由于它是一种周期性的任务，所以最终系统里的 `MessageQueue` 会被分配完全。

`RocketMQ` 的 `Rebalance` 机制依靠客户端各自单独计算得到，`Kafka` 新版本中则依靠 `Consumer Leader` 单点计算后再上传至 `Group Coordinator`，由它下发至每个消费者实例进行更新。

这两种方式各有优缺点，通常来说，单点计算可以最大程度减小视图不一致导致的频繁 `Rebalance` 现象（但也不能杜绝），但是缺点在于逻辑复杂，消费者组和 `Broker` 中都需要选取单点，一个负责计算一个负责下发通知；客户端计算实现上更简单，彼此独立，通过周期性任务最终也能完成重新分配的任务，但是由于客户端彼此获取的视图不做校验，因此可能存在由于视图不一致导致的重复消费和频繁 `Rebalance`。

### RocketMQ 消息存储 ###

硬核内容很多，而且文件存储我接触的也不多更不敢瞎写了，这块后续会视我的学习情况看看是否单独再开一个坑。

如果有同学很想了解这部分内容的话，我贴几篇在资料搜集过程中看到的比较好的博文：

1.  RocketMQ高性能之底层存储设计
2.  RocketMQ 消息存储流程

### 总结 ###

如果你把 `RocketMQ` 和 `Kafka` 对比起来看，其实消息队列的设计哲学有很多相似之处，但在文件存储粒度、分区容灾、负载均衡等方面，二者又有自己的设计考量，采用了不同的实现思路，结合 `Kafka` 的 `ISR` 同步、`Rebalance`、`Partition Failover` 等机制一起学习的话，这种感受会更强烈一些，希望这篇文章对大家有所启发。

[ca57a04c9932ba80763952c4aa316fa1.png]: /images/20221119/78eee7eb96dc4b64a161bb63589bf0ba.png
[3218bcba81ddd2644643ac3a88948563.png]: /images/20221119/eef9081a5e3b446983bd4ae9ea6e6bae.png
[928cbb8f762af35740161b3c2fa54357.png]: /images/20221119/3bc4512483904204bae5b75ccaae5c5f.png
[1df14cdc8b830b8df33dab08abd641a1.png]: /images/20221119/3569e2018d714e84849c6c7959e7c89d.png
[74f6e1eec17649407716d3b827ae557a.png]: /images/20221119/4011e2c52a2c4616b55401332129d4f3.png
[dba9484946266f630cdbfed896f79bf7.png]: /images/20221119/a12bb87b59114c4fa60a1d6fcf320558.png
[40c798062228f30111daac2f6dd2b43b.png]: /images/20221119/f94c852467994ee0aadc495f2a3fac88.png
[78f6345ed895b9cfa08570a79f117b00.png]: /images/20221119/22535f3e517c4e9492c605ffe7e24272.png
[ddcab2bae82c0f796d80371faeb3977f.png]: /images/20221119/770cbec34487488691c7888540caa5d1.png
[8ca0c7f4ff9e2807cb72f1f4d5a2324f.png]: /images/20221119/91d4f2f1c20e4de7b40aa45e7630d01b.png