消息队列系列（二）关于RocketMQ的内部构造设计剖析

分析RocketMQ内部的Broker设计

下边我们来着重分析下Broker节点内部的存储设计思路。

当消息投递到Broker节点内部之后，如果让我们来进行消息存储的设计，你会如何构思呢，下边给出一些我个人的思考：

Version1 单一队列存放模式

所有的消息都存放在一个统一的队列中，并且该队列的内部存储形式是以文件持久化的方式写在IO中。这块实际应用到代码层面的设计可以大概如下：

class MsgInfo{
    byte[] data;
    int msgSize;
    MsgInfo pre;
    MsgInfo next;
}

思考：消费方通常会有多个，如果将所有的消息都杂糅到了一个队列中，那么各个消费者该如何去读取各自的消息呢？

引出Version2的设计

Version2 二级队列的设计

设计一个类似索引一样的目录，为每种消息都单独创建一条队列，然后消费者针对自己希望获得的消息对不同的队列进行监听即可。例如以下这种设计：

思考：队列内部需要存储实际的消息吗？

不需要，只需要存储消息的实际存储地址，然后根据地址定位到CommitLog的具体位置即可。

思考：上边的这种设计存在什么缺陷？

队列内部存储消息在CommitLog的实际地址，当消息频繁的时候，CommitLog会存在大量随机读的情况发生。

思考：应对每种消息类型都只有一条队列，如果遇到高并发场景，消费速率可出现跟不上生产速率该如何解决？

引出Version3的设计

Version3 多二级队列+多消费者设计

在原先的单一队列基础上做扩充，例如Topic-A相关的消息，按照一定的路由算法将它分散到多个不同队列，然后不同的队列都交给多台consumer节点的机器去分别消费。

思考：假设我们分配了3条队列，但是如果只有2个消费者，那么该如何分配呢？

这个时候可以根据机器的性能进行评估，其中1个消费者负责2条队列，另一个消费者只负责1条队列的消费。保证1个消费者可以消费多个队列，但是1条队列只能给到1个消费者消费。

思考：**每个队列只给一个消费者去消费的话，该如何记录当前消费的进度呢？**

引出Version4的设计

**Version4 加入游标索引对当前消息的消费进度进行记录**

大概设计思路为：

consumerOffset：当前消费者所消费到的最新消息，紫色表示已经接收到了ACK。

maxOffset：当前队列中存放的最新消息下标位置，图中绿色的消息表示还未收到ACK确认信息。

minOffset：表示接受最小确认消息游标，由于在minOffset和consumeOffset之间存在着未接收到ACK确认信息的消息，所以minOffset会小于consumeOffset，存在消息偏差。只有当minOffset和consumeOffset之间的消息全部都被ACK确认了之后，才会将minOffset偏移量进行挪动。

ps:这里你可能会疑惑，为什么有了consumerOffset之后还需要设计一个minOffset，这是因为同一条消息队列有可能会被多个消费组开展消费工作，不同的消费组对于消息的处理顺序不相同，所以就会设计一个叫做minOffset的变量用于存储当前多个消费组中消费进度最低的那组数据。例如下图所示：

ps：通常位于minOffset之前的消息会定期从broker节点中进行清理，保证broker有足够的磁盘空间存储消息数据。

思考：消费完毕之后，broker如何知道对Offset的调整？

在消费者消费完毕之后需要返回ACK确认机制，在ACK确认信号的数据包中会携带一个nextBeginOffset的标记，告知Broker可以将ConsumeOffset往后进行移动。

思考：发送消息的时候，消息都被存储在了commitLog中，那么如何将其划分到对应的队列里呢？

这里需要加入一个路由的决策机制，通过在发送方加入一个消息路由的决策信息，假设我们为一个消息创建了8条队列，那么可以按照依次轮询的方式将消息发送给不同的队列。关于路由策略的模块，后边我会单独开一篇文章对它展开深入研究。

关于RocketMQ的一些落地使用经验分析

这里我会结合一些实战场景和大家介绍下这些知识点的实际应用领域。

如何保证消息的顺序性

应用具体场景：

业务集群中，有一台机器专门用于做canal-client，对数据库的binlog进行订阅，然后将其转换为消息，发送到rocketMq中，再由各个消费方去接入对应的消息进行消费。

通常消费方会通过订阅MQ的消息来实现对某些数据表的新增记录监听。

思考：binlog的新增后都会发到MQ中，如何确保MQ发送过来的消息顺序是和binlog的顺序完全一致的？

解决这个问题之前，我们需要先明白为什么消息存在顺序不一致的问题。这是由于RocketMQ的消息在经过Producer发送的时候，会被路由策略发送到不同的queue上，而通常在集群环境中，同一个Topic对应的多个Queue都是存储在不同的Broker节点上的，大致存储结构如下图所示：

消息发送出去之后，通过producer的路由策略会被存储在两个Broker节点上，而消费者拉取消息的时候会从两台机器上拉取，一旦出现一些网络信息的抖动就有可能出现两个Broker上的consumeOffset不一致，导致顺序不能实现强一致性。（例如Broker-A节点先拉取3条消息，再从Broker-B节点拉取3条消息）

这种多队列的模式下实现整体有序性还是可以的，但是要做到严格有序性就会比较难了。所以如何解决有序性问题的根本方式就是只允许有一条队列存在，从一端投递，另一端消费。

消息回溯的具体应用

已经确认的消息是否是否可以支持消息重新投递？

答案是可以的。而且这类功能也是在实际开发过程中非常常用的一点。

例如A团队和B团队进行协同对接开发，两边约定好通过消息队列的方式进行对接，A负责发送消息，B负责在下游监听相关消息的内容。

某天，A团队先在测试环境将代码部署好了，但是B团队还需加班开发，此时A可以预先发送一部分消息并且将其存放在Broker中，然后B团队的同学可以i通过消息回溯的方式来对自己的代码进行debug。

消息回溯功能在RocketMQ的控制台上有具体的功能。

如何解决消息堆积问题

我们都知道使用RocketMQ来进行削峰的操作，但是在面对一些突发的高并发场景，削峰可能会造成生产者消息发送过快，消费者能力跟不上的情况，这种场景下就会发生消息堆积的问题了。

其实这个问题可以将它广义化，因为产生消息堆积的根源其实就只有两个——生产者生产太快或者消费者消费太慢。

我们可以从多个角度去思考解决这个问题，当流量到峰值的时候是因为生产者生产太快，我们可以使用一些 限流降级 的方法，当然你也可以增加多个消费者实例去水平扩展增加消费能力来匹配生产的激增。如果消费者消费过慢的话，我们可以先检查 是否是消费者出现了大量的消费错误 ，或者打印一下日志查看是否是哪一个线程卡死，出现了锁资源不释放等等的问题。

当然，最快速解决消息堆积问题的方法还是增加消费者实例，不过 同时你还需要增加每个主题的队列数量 。

别忘了在 RocketMQ 中，一个队列只会被一个消费者消费 ，如果你仅仅是增加消费者实例的话，如果消费者数目大于实际队列数目，那么就会出现消费者获取不到消息的情况，这个时候就需要对队列进行拷贝拆分的操作了。

消息的PUSH模式实现原理

注意，RocketMQ会存在有低延迟问题，其中就包括这个消息的 push 延迟问题。因为这并不是真正的将消息主动的推送到消费者，而是 Broker 定时任务每5s将消息推送到消费者。

pull模式需要我们手动调用consumer拉消息，而push模式则只需要我们提供一个listener即可实现对消息的监听，而实际上，RocketMQ的push模式是基于pull模式实现的，它没有实现真正的push。

push方式里，consumer把轮询过程封装了，并注册MessageListener监听器，取到消息后，唤醒MessageListener的consumeMessage()来消费，对用户而言，感觉消息是被推送过来的。

本文由哈喽比特于2年以前收录，如有侵权请联系我们。
文章来源：https://mp.weixin.qq.com/s/1bo-upqPvPK-PXG0gjXReQ