RocketMQ入门笔记

我会带着你远行 2022-10-17 05:53 358阅读 0赞

### 文章目录 ###

*  简介
 *  1.RocketMQ架构组成部分
 *   *  1.1 NameServer
     *  1.2 BrokerServer
     *  1.3 Producer
     *  1.4 Consumer
 *  2.NameServer职责描述
 *   *  2.1 状态维护逻辑
     *  2.2 为何不用Zookeeper
     *  2.3 底层通信
 *  3.消息队列的核心机制
 *   *  3.1消息存储和发送
     *  3.2 刷盘方式
     *  3.3 同步方式
 *  4.消息可靠性
 *   *  4.1 顺序消息
     *  4.2 消息重复
     *  4.3 消息优先级
 *  5.Consumer的负载均衡
 *   *  5.1 DefaultMQPushConsumer的负载均衡
     *  5.2 DefaultMQPullConsumer的负载均衡
 *  6.运维工具
 *   *  6.1 RocketMQ-Console
     *  6.2 基于Tools模块自定义开发

# 简介 #

RocketMQ已经在公司项目中用了一年多了,基本上没出现过问题,抽空把之前的笔记整理下分享出来

RocketMQ是由阿里捐赠给Apache的一款分布式、队列模型的开源消息中间件，经历了淘宝双十一的洗礼。  
[github地址][github]

RocketMQ优点如下:

*  削峰填谷（主要解决瞬时写压力大于应用服务能力导致消息丢失、系统奔溃等问题）
 *  系统解耦（解决不同重要程度、不同能力级别系统之间依赖导致一死全死）
 *  提升性能（当存在一对多调用时，可以发一条消息给消息系统，让消息系统通知相关系统）
 *  蓄流压测（线上有些链路不好压测，可以通过堆积一定量消息再放开来压测）  
    其它特性请参考官方介绍: [rocketmq特性][rocketmq]

部署应用架构图如下:  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70]

# 1.RocketMQ架构组成部分 #

## 1.1 NameServer ##

NameServer是一个非常简单的Topic路由**注册中心**，其角色类似Dubbo中的zookeeper，支持Broker的动态注册与发现。主要包括两个功能：Broker管理，NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活；路由信息管理，每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费。NameServer通常也是集群的方式部署，各实例间相互不进行信息通讯。Broker是向每一台NameServer注册自己的路由信息，所以每一个NameServer实例上面都保存一份完整的路由信息。当某个NameServer因某种原因下线了，Broker仍然可以向其它NameServer同步其路由信息，Producer,Consumer仍然可以动态感知Broker的路由的信息。

## 1.2 BrokerServer ##

Broker主要负责消息的存储、投递和查询以及服务高可用保证，为了实现这些功能，Broker包含了以下几个重要子模块。

*  Remoting Module：整个Broker的实体，负责处理来自clients端的请求。
 *  Client Manager：负责管理客户端(Producer/Consumer)和维护Consumer的Topic订阅信息
 *  Store Service：提供方便简单的API接口处理消息存储到物理硬盘和查询功能。
 *  HA Service：高可用服务，提供Master Broker 和 Slave Broker之间的数据同步功能。
 *  Index Service：根据特定的Message key对投递到Broker的消息进行索引服务，以提供消息的快速查询。

## 1.3 Producer ##

消息发布的角色，支持分布式集群方式部署。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟。

## 1.4 Consumer ##

消息消费的角色，支持分布式集群方式部署。支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。

# 2.NameServer职责描述 #

namesrv主要功能在rocketmq源码中namesrv模块中**RouteInfoManager**，集群的状态就保存以下五个变量中  
![在这里插入图片描述][20210602124701302.png]

*  **topicQueueTable**  
    它存储了所有的topic属性信息,Value是个QueueData队列,队列的长度等于这个Topic数据存储的Master Broker个数,QueueData里存储着Broker的名称,读写queue的数量,同步标识等
 *  **brokerAddrTable**  
    以BrokerName为索引,相同名称的Broker可能存在多台机器,一个Master和多个Slave,这个结构存储着一个BrokerName对应属性信息,包括所属的Cluster名称，一个Master  
    Broker和多个Slave Broker的地址
 *  **clusterAddrTable**  
    存储的是集群中Cluster的信息,结构很简单,就是一个Cluster名称对应一个由BrokerName组成的集合
 *  **brokerLiveTable**  
    这个结构和BrokerAddrTable有关系,但是内容完全不同,这个结构的key是BrokerAddr，也就对应着一台机器,BrokerAddrTable中的Key是BrokerName，多个机器的BrokerName可以相同,BrokerLiveTable存储的内容是这台Broker机器的实时状态,包括上次更新装的时间戳,NameServer会定期检查这个时间戳,超时没有更新就任务Broker无效,将其重Broker列表中清除
 *  **filterServerTable**  
    Filter Server是过滤服务器,是RocketMQ的一种服务端过滤方式,一个Broker可用有一个或者多个Filter Server， 这个结构的Key是Broker的地址,Value是和Broker关联的多个Filter Server的地址

## 2.1 状态维护逻辑 ##

具体逻辑在**BrokerHousekeepingService**类中,当NameServer和Broker的长链接断开以后会调用onChannelDestory函数,把这个Broker的信息清理出去。  
NameServer还有定时检查时间戳的逻辑,Broker向NameServer发送心跳会更新时间戳,当NameServer检查到时间戳长时间没有更新，便会触发清理逻辑  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 1]

## 2.2 为何不用Zookeeper ##

zooKeeper是Apache的一个开源软件,为分布式应用程序提供协调服务,那为什么RocketMQ要自己造轮子,开发集群管理程序呢？答案是ZooKeeper的功能很强大,包括自动Master选举,当RocketMQ的架构设计决定它不需要Master选举等复杂功能,只需要一个轻量级的元数据服务器就足以

## 2.3 底层通信 ##

RocketMQ的通信相关代码在**remoting**模块里，是基于Netty实现通信的,先来看看主要类结构,  
RemotingClient和RemotingServer继承RemotingService接口,并增加自己特有的方法,具体实现类NettyRemotingClinet和NettyRemotingServer分别实现了RemotingClinet和RemotingServer,而且都继承了NettyRemotingAbstract类，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 2]  
![在这里插入图片描述][20210602125928564.png]

通过上面的封装,RocketMQ各个模块间的通信,可以通过发送统一格式的自定义消息(RemotingCommand)来完成,各个模块间的通信实现简洁明了，  
在**NameServerController**里注册**DefaultRequestProcessor**实例调用**processRequest**方法处理请求

下图是源码中的启动加载流程

首先在NamesrvStartup的main函数中实例化**NamesrvController**对象,然后执行NamesrvController对象的initialize()方法,在initialize方法中调用**registerDefaultProcessor**注册DefaultRequestProcessor类。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 3]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 6]

# 3.消息队列的核心机制 #

## 3.1消息存储和发送 ##

RocketMQ利用java7中的MappedByteBuffer特性,也就是[零拷贝技术][Link 1],提高消息存盘和网络发送时间

RocketMQ消息的存储是由**CommitLog**和**CommitQueue**配合完成的,消息真正的物理存储文件是CommitLog，ComsumeQueue是消息的逻辑队列,类似数据库的索引文件,存储的是指向物理存储的地址,每个topic下的每个Message Queue都有一个对应的ConsumeQueue文件,文件的地址在\{user.home\}\\store\\consumequeue\{topicName\}\{queueId\}\{fileName\}  
![在这里插入图片描述][2021060219401146.png]  
![在这里插入图片描述][20210602194215324.png]  
![在这里插入图片描述][20210602194227662.png]

CommitLog存储地址在\{user.home\}\\store\\commitlog\{fileName\]，

![在这里插入图片描述][20210602194411878.png]

RocketMQ采取一些机制,尽量向CommitLog中顺序写,但是随机读.

存储机制这样设计的好处

*  CommitLog顺序写,可以大大提高写入效率
 *  虽然是随机读,但是利用操作系统的pagecache机制,可以批量地从磁盘读取,作为cache存到内存中,加快后续的读取速度。
 *  ConsumeQueue存储的偏移量信息，此外为了保证CommitLog和ConsumeQueue的一致性,CommitLog里存储了Consume Queue，Message Key,Tag等所有信息，即使ComsumeQueue丢失,也可以通过commitLog完全恢复

## 3.2 刷盘方式 ##

通过Broker配置文件里的**fiushDiskType**参数设置的，(异步刷盘)ASYNC\_FLUSH和(同步刷盘)SYNC\_FLUSH

*  **同步刷盘:** 发送消息时在返回写成功状态时,消息已经被写入磁盘,具体流程是,消息写入内存的PAGECACHE后,立即通知刷盘线程刷盘,然后等待刷盘完成,刷盘线程执行完成后唤醒等待的显存,返回消息写成功.
 *  **异步刷盘:** 发送消息时在返回写成功状态时,消息可能只是被写入了内存的PAGECACHE,写操作的返回快,吞吐量大;当内存的消息累计到达一定程度时,统一触发写磁盘动作,快速写入.

## 3.3 同步方式 ##

通过Broker配置文件中的**brokerRole**参数设置同步复制还是异步 ASYNC\_MASTER,SYNC\_MASTER,SLAVE三个值中的一个 如果一个Broker组有Master和Slave,消息需要从Master复制到Slave上,有同步和异步两种复制方式。

*  **同步复制:** 发送消息时等Master和Slave均写入成功才反馈给客户端写入成功状态,如果Master出现故障,Slave上有全部的备份数据,容易恢复，但是同步复制会增大数据写入延迟,降低系统的吞吐量
 *  **异步复制:** 发送消息时等Master写入成功即可反馈给客户端写入成功状态,系统拥有较低的延迟和较高的吞吐量,但是Master出了故障,有些数据因为没有被写入Slave，有可能会丢失.

# 4.消息可靠性 #

## 4.1 顺序消息 ##

*  **全局顺序消息:** 要保证全局顺序消息,需要把topic的读写队列数设置为一,然后Product和Consumer的并发设置也要是一
 *  **部分顺序消息:** 要保证部分消息有序,需要发送端和消费端配合处理,在发送端,要做到把同一业务ID的消息发送到同一个Message Queue;在消费过程中,要做到从同一个Message Queue读取的消息不能被并发处理，这样才能达到部分有序

## 4.2 消息重复 ##

对于分布式消息队列来说，同时做到确保一定投递和不重复投递是很难的,也就是所谓的 “一次成功”,在鱼和熊掌不可兼得的情况下,RocketMQ选择了确保一定投递,保证消息不丢失,但是可能会造成消息重复  
解决方法有以下两种

*  保证消费逻辑的幂等性(多次调用和一次调用效果相同)
 *  维护一个已消费消息的记录,消费前查询这个消息是否被消费过

下图是配置消息发送失败重试的相关参数  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 7]

## 4.3 消息优先级 ##

RocketMQ是一个先进先出的队列,不支持消息级别或者Topic级别的优先级,  
业务中简单的优先级需求可以通过间接的方式实现

以下列出三种场景

*  1.当前topic里有太多相似的消息,例如A,B,C,A是优先级高需要处理的，可以把A单独放一个队列中,B和C放入一个,应用程序创建2个Consumer分别订阅不同的topic
 *  2.订单处理系统需要接受100家快递门店过来的请求,把这些请求通过Product写入RocketMQ,订单处理程序每天只能处理1万单,如果其中1个门店一天发送了2万单,这样其他的99家门店消息可能被迫等待2万单处理完后才能处理,显然很不公平  
    解决方法: 创建一个Topic,设置Topic的**MessageQueue**数量超过100个,Product根据订单的门店号来决定消息写入哪个MessageQueue,DefaultMQPushConsumer默认是采用循环的方式读取一个Topic的所有MessageQueue,这样如果某家订单量大增也不会影响其他门店使用  
    DefaultMQPushConsumer默认的**pullBatchSize**是32,也就是每次重某个MessageQueue读取消息的时候,最多读32个，在上面场景中可以把pullBatchSize设置为1
 *  3 强优先级场景，上面两种情况对优先级要求不高,更像一个保证公平的机制，如果消息T1>T2>T3这种应用场景,需要用户自己使用pullConsumer控制消息的消费

# 5.Consumer的负载均衡 #

## 5.1 DefaultMQPushConsumer的负载均衡 ##

负载均衡算法有五种,

*  平均分配策略(默认)(AllocateMessageQueueAveragely)
 *  环形分配策略(AllocateMessageQueueAveragelyByCircle)
 *  手动配置分配策略(AllocateMessageQueueByConfig)
 *  机房分配策略(AllocateMessageQueueByMachineRoom)
 *  一致性哈希分配策略(AllocateMessageQueueConsistentHash)

默认用第一种A\*\*llocateMessageQueueAveragely,\*\*负载均衡的结果和Topic的Message Queue数量,以及ConsumerGroup里的Consumer的数量有关系,当Message Queue书设置为3,当Consumer数量为2的时候,有一个Consumer需要处理2/3的请求,如果当Consumer数量为4的时候,有一个Consumer不会接受到请求  
![在这里插入图片描述][20210602200919561.png]  
![在这里插入图片描述][20210602201656186.png]

## 5.2 DefaultMQPullConsumer的负载均衡 ##

PullConsumer可以看到所有的Message Queue,而且重哪个Message Queue读取消息,读消息时的Offset都由使用者控制,使用者可以实现任何方式的负载均衡

# 6.运维工具 #

## 6.1 RocketMQ-Console ##

是一个基于spring boot开发的运维页面工具，我看可以参考它的源码进行自定义运维工具的开发  
[github地址][github 1]

## 6.2 基于Tools模块自定义开发 ##

RocketMQ源码中有一个tools模块,MQAdmin相关命令的实现就在这里,Tools模块源码中有一个command包,里面列出了各个组件相关的命令

参考文章:  
书籍: **RokcetMQ实战与原理解析**  
RocketMQ官网文档: [https://github.com/apache/rocketmq/tree/master/docs/cn][https_github.com_apache_rocketmq_tree_master_docs_cn]

[github]: https://github.com/apache/rocketmq
[rocketmq]: https://github.com/apache/rocketmq/blob/master/docs/cn/features.md
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70]: /images/20221014/0119d9fa455e47079d5f2a3f7d350d80.png
[20210602124701302.png]: /images/20221014/a216d67b6cb3469b98b8d380fb2eb96a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 1]: /images/20221014/d7e242ef9296408f85d2339b001d06fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 2]: /images/20221014/94d7b26f9b8a46e7959f82625af71f69.png
[20210602125928564.png]: /images/20221014/140d86d8735f489da2851c17c2f0fdc0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 3]: /images/20221014/c7fe84dcd18141eca7913af029c2d9b4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 4]: /images/20221014/a2a0dfddcad74d37b9b05b8b817f91d6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 5]: /images/20221014/48dfce30679840e8a43eb440314d4509.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 6]: /images/20221014/9ac6d469ad2a4e3b9870a3c2154e8ace.png
[Link 1]: https://www.jianshu.com/p/497e7640b57c
[2021060219401146.png]: /images/20221014/400cfa0c66b043c5820edf364090a883.png
[20210602194215324.png]: /images/20221014/d04cfb27213a401a978c2938b9c20fde.png
[20210602194227662.png]: /images/20221014/b9d3d174220d43649ef9f549b76b954e.png
[20210602194411878.png]: /images/20221014/92be82e8706643678dcc130c25f12df8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L21pbmcxOTk1MTIyNA_size_16_color_FFFFFF_t_70 7]: /images/20221014/c52797c08a0d4e3eb66f362a63c7a9fa.png
[20210602200919561.png]: /images/20221014/f9749607defe4fc38e7f3fa00a8e4906.png
[20210602201656186.png]: /images/20221014/7f25238680014f709ce376acdb34c345.png
[github 1]: https://github.com/apache/rocketmq-externals/tree/master/rocketmq-console
[https_github.com_apache_rocketmq_tree_master_docs_cn]: https://github.com/apache/rocketmq/tree/master/docs/cn