RocketMQ第一篇：入门介绍及启动部分源码分析

一、RocketMQ简介

1.1、介绍：

RocketMQ是一款分布式、队列模型的消息中间件，由Metaq3.X版本改名而来，RocketMQ并不遵循包括JMS规范在内的任何规范，但是参考了各种规范不同类产品的设计思想，自己有一套自定义的机制，简单来说就是使用订阅主题的方式去发送和接收任务，但是支持集群和广播两种消息模式。
开源项目地址：https://github.com/apache/rocketmq

1.2、特点：

1.2.1、削峰填谷

主要解决诸如秒杀、抢红包、企业开门红等大型活动时皆会带来较高的流量脉冲，或因没做相应的保护而导致系统超负荷甚至崩溃，或因限制太过导致请求大量失败而影响用户体验，亿级消息堆积能力，消息堆积能力强。

1.2.2、异步解耦

高可用松耦合架构设计，对高依赖的项目之间进行解耦，当下游系统出现宕机，不会影响上游系统的正常运行，或者雪崩。

1.2.3、顺序消息

顺序消息即保证消息的先进先出，比如证券交易过程时间优先原则，交易系统中的订单创建、支付、退款等流程，航班中的旅客登机消息处理等。

1.2.4、分布式事务消息

确保数据的最终一致性，大量引入 MQ 的分布式事务，既可以实现系统之间的解耦，又可以保证最终的数据一致性，减少系统间的交互。

1.3、选用理由：

1、强调集群无单点，可扩展，任意一点高可用，水平可扩展。
2、海量消息堆积能力，消息堆积后，写入低延迟。
3、支持上万个队列。
4、消息失败重试机制。
5、开源社区活跃。
6、成熟度高（历经多次天猫双十一海量消息考验）

二、RocketMQ安装

网上很多相关文章，不再赘述

三、Rocket MQ架构分析

3.1、RocketMQ 整体的架构设计

分别介绍一下上面的几个主要概念：

3.1.1、Producer

消息生产者，负责生产消息，一般由业务系统负责生产消息。
一个消息生产者会把业务应用系统里产生的消息发送到Broker 服务器。
RocketMQ 提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。
同步和异步方式均需要 Broker 返回确认信息，单向发送不需要。

3.1.2、Consumer

负责消费消息，一般是后台系统负责异步消费。
一个消息消费者会从 Broker 服务器拉取消息、并将其提供给应用程序。
从用户应用的角度而言提供了两种消费形式：拉取式消费、推动式消费。

3.1.4、Producer Cluster（Group）

一类 Producer 的集合名称，这类 Producer 通常发送一类消息，且发送逻辑一致。

3.1.5、Consumer Cluster（Group）

一类 Consumer 的集合名称，这类 Consumer 通常消费一类消息，且消费逻辑一致。

3.1.6、NameServer

名称服务充当路由消息的提供者。
生产者或消费者能够通过名字服务查找各主题（Topic）相应的 Broker IP 列表。
多个 Namesrver 实例组成集群，但相互独立，没有信息交换。

3.1.7、BrokerServer

消息中转角色，负责存储消息、转发消息。
代理服务器在 RocketMQ 系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。
在 JMS 规范中称为 Provider。

3.1.8、Message

消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。
RocketMQ 中每个消息拥有唯一的Message ID，且可以携带具有业务标识的Key。
系统提供了通过Message ID和Key查询消息的功能。

3.1.9、Topic

表示一类消息的集合，每个主题包含若干条消息，每条消息只能属于一个主题，是 RocketMQ 进行消息订阅的基本单位。

3.1.10、Tag

为消息设置的标志，用于同一主题下区分不同类型的消息。
可以理解为 Topic 的二级分类。
来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。
标签能够有效地保持代码的清晰度和连贯性，并优化 RocketMQ 提供的查询系统。
消费者可以根据 Tag 实现对不同子主题的不同消费逻辑，实现更好的扩展性。

3.1.11、消费模式

消息消费模式有两种：Clustering（集群消费）和 Broadcasting（广播消费）

默认是 Clustering 模式，该模式下同一个消费者集群中订阅一个主题，一个消息只会被一个消费者消费。
类似于打移动客服电话，只会被一个客服接听。

而 Broadcasting 模式消息会发给消费者组中的每一个消费者进行消费。

3.1.12、消费方式

Consumer 端有两种消费形式：Pull（拉取式消费）、Push（推动式消费）

拉取式消费（Pull Consumer）：主动调用 Consumer 的拉消息方法从 Broker 服务器拉消息、主动权由应用控制。
一旦获取了批量消息，应用就会启动消费过程。

推动式消费（Push Consumer）：模式下 Broker 收到数据后会主动推送给消费端，该消费模式一般实时性较高。
应用通常向 Consumer 对象注册一个 Listener 接口，一旦收到消息，Consumer 即立刻回调 Listener 接口方法
（其实 Push 模式只是对 Pull 模式的一种封装，其本质实现为消息拉取线程在从服务器拉取到一批消息后，然后提交到消息消费线程池后，又“马不停蹄”的继续向服务器再次尝试拉取消息）

3.1.13、消息类型

(1)、消息类型有两种：

Normal Message（普通消息） 和 Ordered Message（顺序消费）。
顺序消费是指消息的消费顺序能够与消息的发送顺序一致。
但是有时候我们从业务需要上面并不需要保证所有消息严格按照消费顺序完全一致。
例如，一个订单的下单、付款、出库等操作是不同替换顺序。
但是有A订单和B订单，并不需要保证A订单与B订单的顺序。

RocketMQ采用了局部顺序一致性的机制，一组消息发送到同一个队列中来保证发送顺序的有序性，然后再由消费者进行消费。

消费的时候通过一个队列只会被一个线程取到 ，第二个线程无法访问这个队列 来保证队列有序性。

RocketMQ可以同时多个队列并列消费，提高rocketmq的消费速度。

其实这个方案很像jdk7 里面ConcurrentHashMap 实现分段锁的实现，通过保证每段的线程安全，多段并行消费提高消费能力。

(2)、顺序消费分为普通顺序、严格顺序。

(2.1)、普通顺序消费模式下，消费者通过同一个消费队列收到的消息是有顺序的，不同消息队列收到的消息则可能是无顺序的。

普通顺序消息在 Broker 重启情况下不会保证消息顺序性 (短暂时间) 。

(2.2)、严格顺序消息模式下，消费者收到的所有消息均是有顺序的。

严格顺序消息模式下，对于指定的一个 Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。严格顺序消息 即使在异常情况下也会保证消息的顺序性 。
严格顺序虽然能更好的保证消息有序，但实现它可会付出巨大的代价。如果你使用严格顺序模式，Broker 集群中只要有一台机器不可用，则整个集群都不可用。
一般而言，我们的 MQ 都是能容忍短暂的乱序，所以推荐使用普通顺序模式。

(3)、顺序消费的实现：

在MQ的模型中，顺序需要由3个阶段去保障：
1.消息被发送时保持顺序
2.消息被存储时保持和发送的顺序一致
3.消息被消费时保持和存储的顺序一致

(3.1)、消息被发送时保持顺序

(3.1.1)、使用严格顺序模式

严格顺序消息模式下，对于指定的一个 Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。
因此只要保证消息同步发送（发完一条后再发下一条）即可保证消息发送时保持顺序。

(3.1.2)、使用普通顺序模式

普通顺序模式下，只有同一个队列的消息能保证有序。
Producer 生产消息的时候会进行轮询(根据设定的负载均衡策略)来向同一主题的不同消息队列发送消息。
那么如果此时有几个消息分别是同一个订单的创建、支付、发货，在轮询的策略下这 三个消息会被发送到不同队列 ，因为在不同的队列，此时就无法使用 RocketMQ 带来的队列有序特性来保证消息有序性了。

因此使用普通顺序时，在同步发送的基础上，还需要将消息发送到相同的队列。
在RocketMQ中，通过MessageQueueSelector来实现队列的选择。通过对订单的唯一标识符取hash，将同一个订单的消息发送到相同的队列。

(3.2)、消息被消费时保持和存储的顺序一致

在分布式的情况下，即使消息队列有序的将消息发送给消费者，也可能因为网络等原因，导致消费者接收到的消息无序。
如：按顺序发送消息a、b给消费者。虽然a先发送，但因为网络原因，消息a在网络中滞留一段时间，导致消费者收到的消息顺序为b、a。同时，若同一个队列的消息由不同消费者消费也可能出现以上情况。

(3.3)、消费者顺序消费消息的实现

基于以上分析，要保证消息顺序的被消费者消费，必须满足下列条件：

(3.3.1)、同一个订单的消息由同一个消费者消费

(3.3.2)、消费者消费完一条消息之后，才可以接着消费下一条

（1）同一个消费者消费
类似于通过订单id的hash选择相同的队列，可以通过订单的hash选择同一个消费者同步消费（消费完一条后再拉取下一条，单线程消费），保证同一个订单的顺序消费。

（2）通过 consumer 内部用内存队列做排队，然后分发给底层不同的 worker 实现（实现复杂）。
若消费者是多线程，此时在消费者内部建立内存队列。
先将消息拉取到内存队列后，在分发给不同的线程。

3.2、源码目录结构

3.3、各角色之间具体的交互流程

重点了解三个概念:

（1）、CommitLog：

消息存储的主体结构，简单来说就是存储Producer发送的消息。要知道所有的消息都是需要落盘的，所以这些消息都是要写入文件。
所写的每个文件默认大小为1G（为什么默认为1G，涉及到页缓存与内存映射的概念） ，文件满了写入下一个文件。
页缓存与内存映射
PageCache（页缓存）：
文件系统缓存，加速文件的读写速度。我们都知道磁盘 IO 和内存 IO 的速度可是相差了好几个数量级。加入页缓存的目的就是为了使程序对文件进行顺序读写的速度接近于内存的读写速度。所以简单来说，对于数据的写入，OS 会先写入至 Cache 内，随后通过异步的方式由 pdflush 内核线程将 Cache 内的数据刷盘至物理磁盘上。对于数据的读取，如果一次读取文件时出现未命中 PageCache 的情况，OS 从物理磁盘上访问读取文件的同时，会顺序对其他相邻块的数据文件进行预读取。
MMAP（内存映射）：
将一个文件或者其它对象映射到进程的地址空间，实现文件磁盘地址和进程虚拟地址空间中一段虚拟地址的一一对映关系。简单来说，就是实现磁盘文件到虚拟内存的直接传输，减少了内核态到用户态的数据拷贝。
另外需要说明的是， MMAP 技术在进行文件映射的时候，一般有大小限制，在 1.5GB~2GB 之间。所以 RocketMQ 才让CommitLog单个文件在1GB，ConsumeQueue文件在5.72MB，不会太大。
更详细的解释可以参考文章：https://blog.csdn.net/qfzhangwei/article/details/102531340

（2）、ConsumeQueue：

消息消费队列，可以理解为基于 Topic 的 commitlog 索引文件。Topic 上的 Queue 与 消费者的 ConsumeQueue 一一对应，比如 Topic 有 1，2，3，4个队列，消费者A分配到1，2两个队列（此处涉及消费者负载均衡）,那么消费者A的ConsumerQueue就是对应 Topic1，2 的两个queue。
引入ConsumeQueue 主要是为了提高消息消费的性能，它存储了起始物理偏移量 offset，消息大小 size 和消息 Tag 的 HashCode 值。

（3）、IndexFile：索引文件，简单说就是 commitLog 的索引集合文件。
固定的单个 IndexFile 文件大小约为400M，一个IndexFile 可以保存 2000W个索引，IndexFile的底层存储设计为在文件系统中实现 HashMap 结构，故rocketmq 的索引文件其底层实现为 hash 索引。

3.4、消息刷盘

同步刷盘：
如上图所示，只有在消息真正持久化至磁盘后 RocketMQ 的 Broker 端才会真正返回给 Producer 端一个成功的 ACK 响应。
同步刷盘对 MQ 消息可靠性来说是一种不错的保障，但是性能上会有较大影响，一般适用于金融业务应用该模式较多。
异步刷盘：
能够充分利用 OS 的 PageCache 的优势，只要消息写入 PageCache 即可将成功的 ACK 返回给 Producer 端。
消息刷盘采用后台异步线程提交的方式进行，降低了读写延迟，提高了 MQ 的性能和吞吐量。

3.5、消息重复

RocketMQ不解决消息重复问题，由用户通过外置全局存储自己做判重处理。

RocketMQ单机写入TPS单实例约7万条/秒，单机部署3个Broker，可以跑到最高12万条/秒，消息大小10个字节。换句话说单机RocketMQ的每分钟处理的请求是12W60=720W，每小时处理的请求是720W60=44400W。在这么大的信息量面前任何额外的要求都有点吹毛求疵。

而消息的传递具有不可靠性。
网络不可靠性：只要通过网络传输的消息都具有网络不可靠性；或者说系统受到黑客的恶意篡改，导致的消息完全一致等等。只要消息经过传递，希望在传递层保证消息都无法100%保证消息的可靠性。传递过程无法确保消息不重复，那么消息源也就不需要关注了，因为即使消息源确保唯一，传递过程中还是会产生重复消息。

   消息流从五个环节从消息源，消息发送，消息传递，消息消费都无法保证消息不重复，那么我们能做的只有在消息持久化环节保证消息不重复。其实所有的保证消息不重复的策略都需要一个消息持久化的位置供消息重复验证，然而不巧的是除非和消息最本源的位置做验证，其他环节的验证都具有不可靠性。

消息持久层做消息唯一性的策略：

1.

持久化过程中业务唯一标识验证，每个消息具有业务唯一标识，在消息最终持久化之前通过验证唯一性标识保证消息的唯一性。消息持久化位置如果出现同样的消息，系统就不做处理，期间无任何传递过程，保证消息的唯一性。

2.

使用过程中业务唯一标识验证，使用过程中如果出现同样的消息，系统进行相应的异常处理。
eg: mysql去重表

3.6、事务消息

Apache RocketMQ 在 4.3.0 版中已经支持分布式事务消息，这里 RocketMQ 采用了 2PC 的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息，如下图所示：

Half Message（半消息）
Producer 把消息发送到 Broker 端时，该消息是不能被 Consumer 消费的，需要 Producer 对消息进行二次确认后，才能被消费。

消息回查
由于网络抖动或者 Producer 重启，导致 Producer 一直没有对 Half Message 进行二次确认。Broker 端对未确定状态的消息发起回查，将消息发送到对应的 Producer 端（同一个Group的Producer），由 Producer 根据消息来检查本地事务的状态，进而执行 Commit 或者 Rollback 。值得注意的是，RocketMqQ 并不会无休止的的信息事务状态回查，默认回查15次，如果15次回查还是无法得知事务状态，默认回滚该消息。

回溯消费
回溯消费是指 Consumer 已经消费成功的消息，由于业务上需求需要重新消费，要支持此功能，Broker 在向 Consumer 投递成功消息后，消息仍然需要保留。并且重新消费一般是按照时间维度，例如由于 Consumer 系统故障，恢复后需要重新消费1小时前的数据，那么 Broker 要提供一种机制，可以按照时间维度来回退消费进度。RocketMQ 支持按照时间回溯消费，时间维度精确到毫秒。

定时消息
定时消息（延迟队列）是指消息发送到 broker 后，不会立即被消费，等待特定时间投递给真正的 topic。 broker 有配置项messageDelayLevel，默认值为“1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h”，18个level。可以配置自定义 messageDelayLevel。
注意，messageDelayLevel是 broker 的属性，不属于某个 topic。
发消息时，设置 delayLevel 等级即可：msg.setDelayLevel(level)。

level有以下三种情况：
level == 0，消息为非延迟消息
1<=level<=maxLevel，消息延迟特定时间，例如level1，延迟1s
level > maxLevel，则level maxLevel，例如level==20，延迟2h
定时消息会暂存在名为 SCHEDULE_TOPIC_XXXX 的 topic 中，并根据 delayTimeLevel 存入特定的 queue，queueId = delayTimeLevel – 1，即一个 queue 只存相同延迟的消息，保证具有相同发送延迟的消息能够顺序消费。broker会调度地消费 SCHEDULE_TOPIC_XXXX，将消息写入真实的 topic。

需要注意的是，定时消息会在第一次写入和调度写入真实topic时都会计数，因此发送数量、tps都会变高。

消息重试
Consumer 消费消息其实也是需要一个 确认的操作，即 Comsumer 消费成功后需要返回 Broker 一个确认消息，如果没有返回则 Broker 认为这条消息消费失败，失败后会再重试消费该消息，默认重试16次。

RocketMQ 对于重试消息的处理是先保存至 Topic 名称为SCHEDULE_TOPIC_XXXX的延迟队列中，后台定时任务按照对应的时间进行 Delay 后重新保存至%RETRY%+consumerGroup的重试队列中，而且重试次数越多投递延时就越大。

死信队列
当消息达到最大的重试次数之后（默认16次），若消费依然失败，消息就会被投递到 Topic 名称为 %DLQ%+consumerGroup 的DLQ 死信队列，死信队列用于处理无法被正常消费的消息。

RocketMQ 将这种正常情况下无法被消费的消息称为死信消息（Dead-Letter Message），将存储死信消息的特殊队列称为死信队列（Dead-Letter Queue）。在 RocketMQ 中，可以通过使用 console 控制台对死信队列中的消息进行重发来使得消费者实例再次进行消费。

消息重试更详细的说明可以参考https://blog.csdn.net/weixin_43767015/article/details/121135114

四、namesrv源码分析

4.1、架构设计

4.1.1、消息中间件的设计思路

消息中间件的设计思路一般是基于主题订阅发布机制，即生产者发送某一个主题消息到服务器，消息服务器负责将消息持久化存储，消费者订阅感兴趣的主题，由服务器主动推送到消费者(Push模式)或消费者主动向消息服务器拉取(Pull模式)，从而实现生产者和消费者解耦。

4.2 NameServer 解决了什么

4.2.1 存在的问题

为了增加消息中间件的高可用性，避免消息服务器的单点故障导致整个系统瘫痪，通常会部署多台消息服务器共同承担消息的存储。那么问题随之而来：

消息生产者怎么知道消息要发送到哪台消息服务器呢？
若某一台消息服务器宕机了，消息生产者如何动态感知呢？
NameServer 就是为了解决以上问题设计的

4.2.2 解决方式

NameServer 就像一个寻址表，负责 broker 地址的记录：

Broker 在启动时向所有存活的 NameServer 注册，NameServer 会记录当前这个 Broker 的 IP 地址等相关信息
消息生产者在发送消息前，先从 NameServer 获取 Broker 服务器地址列表，然后根据负载均衡算法从列表中选择一台服务器进行发送消息
NameServer 和每一个注册的 Broekr 都保持长连接，每隔 30s 检测 Broker 是否存活，若检测到 Broker 宕机，则从路由注册表中删除。为了降低 NameServer 实现的复杂度，路由变化不会马上通知消息生产者，而是通过消息发送端的容错起止保证消息发送的可用性
NameServer 的高可用是通过部署多台 NameServer 实现的，但 NameServer 服务器彼此之间不通讯，即各个 NameServer 服务器之间在某一时刻的数据并不完全相同，可是这对消息的发送不会造成任何影响 (因为已记录的 broker 仍能接收消息)

4.3、源码跟踪

PS：源码分析思路：

源码分析需要有清晰的思路，以Broker源码为例：要集合下面的这个图关注几个点：

1.

Broker既然要跟NameServer通信，还要接收生产者消费者的请求，那么它就得有网络服务，所以会集成 NettyServer 和 NettyClient

2.

客户端每次请求Broker过来肯定不能同步阻塞啊，所以就会有线程池来应对各种请求

3.

Broker会有不同的功能，例如高可用，Dleger管理CommitLog,
ConsummeQueue等一些也是需要功能的，所以会有不同的功能对应不同的组件

4.

一个应用总会有一些定时任务存在，比如 定时发送心跳，定时清理，定期落盘等等需要一些定时后台线程池
带着上面的4个问题去跟踪代码，更容易有收获。

4.3.1、启动流程分析

1）、启动配置获取和加载

NamesrvConfig：内部参数配置类
NettyServerConfig：网络通信基于Netty

NamesrvConfig：有一些默认配置

（2）、加载完配置文件后走start方法进行初始化：

（3）、对broker进行扫描和移除

查看初始化方法，在初始化方法中定义了定时器，扫描和移除不活跃的broker
NamesrvController.this.routeInfoManager.scanNotActiveBroker();

删除逻辑：onChannelDestroy

4.3.2、路由分析

NameServer的主要作用是为消息的生产者和消息消费者提供关于主题Topic的路由信息，那么NameServer需要存储路由的基础信息，还要管理Broker节点，包括路由注册、路由删除等。

（1）、路由管理类RouteInfoManager

在创建 NamesrvController 的时候即被创建

记录Topic在哪个队列、broker、集群信息

PS：
topicQueueTable
主要记录了有哪些broker有订阅当前的topic，同时记录了读队列和写队列(均默认4个)
brokerAddrTable
主要记录在 nameserver 上注册了的 broker 的信息，包括名称、所属集群以及集群内的节点地址等

clusterAddrTable
主要记录了集群列表，以及集群中的成员名称

brokerLiveTable
brokerLiveTable 中记录了所有活跃的 broker 地址，最后接收到的心跳时间，连接通道等数据

（2）、路由注册流程

路由注册是通过 Broker 与 NameServer 的心跳功能实现的。

Broker 启动时向所有的 NameSever 发送心跳信息，且每隔 30s 想集群中所有的 NameServer 发送心跳包，NameServer 收到心跳包时会更新 brokerLiveTable 缓存中对应的 broker 信息。

NameServer 每隔 10s 会扫描 brokerLiveTable 表，若连续 2min 内没有收到心跳包，则会将这个 Broker 路由信息移除，并关闭连接。
同时，由于 NameServer 之间是无状态的，这种设计也让新加入的 NameServer 节点能够快速与其他 NameServer 保持数据同步。
相关代码：

1、Broker 定时上报信息

org.apache.rocketmq.broker.BrokerController#start

2、上报信息关键代码：

org.apache.rocketmq.broker.out.BrokerOuterAPI#registerBrokerAll

3、NameSrv 处理心跳包

关键代码：
处理注册broker信息

org/apache/rocketmq/namesrv/processor/DefaultRequestProcessor.java:98

获取锁，并开始写操作

（3）、路由删除流程

路由删除有两个触发点：

a：

NameSrv每隔 10s 会扫描 brokerLiveTable 上的所有 broker 节点，查看上次心跳包与系统时间差是否大于 120s， 若超过则认为这个 broker 不可用，移除这个 broker

b：

Broker 正常关闭的情况下，执行 unregisterBroker 指定，主动删除路由
NameSrv扫描发现核心代码：
org.apache.rocketmq.namesrv.NamesrvController#initialize

在onChannelDestory中会维护各个表数据
org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#scanNotActiveBroker

（4）、路由发现

路由发现并非实时的，即当 Topic 路由 发生变化，NameServer 不会主动推送给客户端，而是由客户端定时拉取主题最新的路由。

五、小结

六、知识点

优雅地关闭线程池
在 start 函数内，可以发现调用了 Runtime.getRuntime().addShutdownHook

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！