rocket mq的工作原理

Love The Way You Lie 2022-06-08 05:10 178阅读 0赞

\#\#一、整体介绍 在RocketMQ里，有以下几个核心的模块：Producer，Consumer，Broker，NameSrv。他们之间的关系如下： [![arch][]][arch 1]

先简单了解一下各个模块的功能，下面会有章节详细介绍各个模块的功能。

*  Producer和Consumer很好理解，顾名思义就是生产者和消费者，生产者负责生产消息，消费者负责消费消息，这2块的逻辑都是由业务使用者定义的。
 *  Broker是RocketMQ的核心，Broker实现了消息的存储、拉取等功能。Broker通常以集群方式启动，并可配置主从，每个Broker上提供对指定topic的服务。理解了Broker的原理，以及和其他服务交互的方式就基本弄懂了整个消息中间件的原理。
 *  NameSrv是一个无状态的名称服务，可以集群部署。所有Broker启动的时候会向NameSrv注册自己的信息。Producer会根据目标topic从NameSrv获取到达指定Broker的路由信息，Consumer同理。

对于Producer端RocketMQ采用了轮询的方式保证了负载均衡，Consumer端通常采用cluster集群方式消费消息，我们可以自己定义消息在消息端的分配方式。另外，MQ还提供了顺序消息的特性，简单了解一下MQ提供的特性即可，具体实现后面章节会进行阐述。

# 源码目录结构介绍&Remoting通信层 #

一：源码目录结构介绍 RocketMQ源码分为以下几个package：

*  `rocketmq-broker`：整个mq的核心，他能够接受producer和consumer的请求，并调用store层服务对消息进行处理。HA服务的基本单元，支持同步双写，异步双写等模式。
 *  `rocketmq-client`：：mq客户端实现，目前官方仅仅开源了java版本的mq客户端，c++，go客户端有社区开源贡献。
 *  `rocketmq-common`：一些模块间通用的功能类，比如一些配置文件、常量。
 *  `rocketmq-example`：官方提供的例子，对典型的功能比如order message，push consumer，pull consumer的用法进行了示范。
 *  `rocketmq-filtersrv`：消息过滤服务，相当于在broker和consumer中间加入了一个filter代理。
 *  `rocketmq-remoting`：基于netty的底层通信实现，所有服务间的交互都基于此模块。
 *  `rocketmq-srvut`：解析命令行的工具类。
 *  rocketmq-store：存储层实现，同时包括了索引服务，高可用HA服务实现。
 *  `rocketmq-tools`：mq集群管理工具，提供了消息查询等功能。

底层基于Netty网络库驱动

# producer  #

### producer 1.启动流程 ###

![SouthEast][]

Producer如何感知要发送消息的broker即brokerAddrTable中的值是怎么获得的，

1.      发送消息的时候指定会指定topic，如果producer集合中没有会根据指定topic到namesrv获取topic发布信息TopicPublishInfo，并放入本地集合
    2.      定时从namesrv更新topic路由信息，

Producer与broker间的心跳

Producer定时发送心跳将producer信息（其实就是procduer的group）定时发送到， brokerAddrTable集合中列出的broker上去
    Producer发送消息只发送到master的broker机器，在通过broker的主从复制机制拷贝到broker的slave上去

### producer 2.如何发送消息 ###

Producer轮询某topic下的所有队列的方式来实现发送方的负载均衡

![SouthEast 1][]

1）  Topic下的所有队列如何理解：

![复制代码][copycode.gif]

比如broker1, broker2, borker3三台broker机器都配置了Topic_A
    
    Broker1 的队列为queue0 , queue1
    
    Broker2 的队列为queue0, queue2, queue3,
    
    Broker3 的队列为queue0
    
    当然一般情况下的broker的配置都是一样的
    
    以上当broker启动的时候注册到namesrv的Topic_A队列为共6个分别为：
    
    broker1_queue0, broker1_queue1,
    
    broker2_queue0, broker2_queue1, broker2_queue2,
    
    broker3_queue0,

![复制代码][copycode.gif]

2）  Producer如何实现轮询队列：

![复制代码][copycode.gif]

Producer从namesrv获取的到Topic_A路由信息TopicPublishInfo
    
               --List<MessageQueue>messageQueueList  //Topic_A的所有的队列
    
               --AtomicIntegersendWhichQueue        //自增整型
    
               方法selectOneMessageQueue方法用来选择一个发送队列
    
                        (++sendWitchQueue)% messageQueueList.size为队列集合的下标
    
                        每次获取queue都会通过sendWhichQueue加一来实现对所有queue的轮询
    
                                如果入参lastBrokerName不为空，代表上次选择的queue发送失败，这次选择应该避开同一个queue

![复制代码][copycode.gif]

3）  Producer发消息系统重试:

![复制代码][copycode.gif]

发送失败后，重试几次retryTimesWhenSendFailed = 2
    
    发送消息超时sendMsgTimeout = 3000
    
    Producer通过selectOneMessageQueue方法获取一个MessagQueue对象
    
               --topic            //Topic_A
    
               --brokerName           //代表发送消息到达的broker
    
               --queueId              //代表发送消息的在指定broker上指定topic下的队列编号
    
    向指定broker的指定topic的指定queue发送消息
    
                       发送失败(1)重试次数不到两次(2)发送此条消息花费时间还没有到3000(毫秒), 换个队列继续发送。

![复制代码][copycode.gif]

producer发送普通消息

![SouthEast 2][]

### producer 3.如何发送顺序消息 ###

![复制代码][copycode.gif]

Rocketmq能够保证消息严格顺序，但是Rocketmq需要producer保证顺序消息按顺序发送到同一个queue中，比如购买流程(1)下单(2)支付(3)支付成功，
    这三个消息需要根据特定规则将这个三个消息按顺序发送到一个queue
    
    如何实现把顺序消息发送到同一个queue:
    
               一般消息是通过轮询所有队列发送的，顺序消息可以根据业务比如说订单号orderId相同的消息发送到同一个队列, 或者同一用户userId发送到同一队列等等
    
    messageQueueList [orderId%messageQueueList.size()]
    
    messageQueueList [userId%messageQueueList.size()]

![复制代码][copycode.gif]

![SouthEast 3][]

### producer 4.如何发布分布式事务消息 ###

先引入官方文档图：

![SouthEast 4][]

分布式事物是基于二阶段提交的

![复制代码][copycode.gif]

1)      一阶段，向broker发送一条prepared的消息，返回消息的offset即消息地址commitLog中消息偏移量。Prepared状态消息不被消费
    发送消息ok，执行本地事物分支， 本地事物方法需要实现rocketmq的回调接口
    
    2) LocalTransactionExecuter，
    处理本地事物逻辑返回处理的事物状态LocalTransactionState
    
    3)  二阶段，处理完本地事物中业务得到事物状态， 根据offset查找到commitLog中的prepared消息，设置消息状态commitType或者rollbackType， 
    让后将信息添加到commitLog中， 其实二阶段生成了两条消息

![复制代码][copycode.gif]

事物消息发送

![SouthEast 5][]

### producer 5.消息在落地broker落地之普通消息 ###

Broker根据producer请求的RequestCode.SEND_MESSAGE选择对应的处理器SendMessageProcessor
    
             根据请求消息内容构建消息内部结构MessageExtBrokerInner
    
             调DefaultMessageStore加消息写入commitlog

![SouthEast 6][]

### producer 6.消息在落地broker落地之事务消息 ###

#### 1. 消息落地 ####

commitLog针对事物消息的处理，消息的第20位开始的八位记录是的消息在逻辑队列中的queueoffset，
    但是针对事物消息为preparedType和rollbackType的存储的是事物状态表的索引偏移量

#### 2. 分发事物消息：    ####

![复制代码][copycode.gif]

分发消息位置信息到ConsumeQueue: 事物状态为preparedType和rollbackType的消息不会将请求分发到ConsumeQueue中去，即不处理，所以不会被消息
      更新transactionstable table：如果是prepared消息记，通过TransactionStateService服务将消息加到存储事务状态的表格tranStateTable的文件中；
      如果是commitType和rollbackType消息， 修改事物状态表格tranStateTable中的消息状态。
      记录Transaction Redo Log日志： 记录了 commitLogOffset, msgSize,preapredTransactionOffset, storeTimestamp。

![复制代码][copycode.gif]

#### 3. 事物状态表 ####

事物状态表是有MapedFileQueue将多个文件组成一个连续的队列，它的存储单元是定长为24个字节的数据，
    
             tranStateTableOffset可以认为是事物状态消息的个数，索引偏移量， 它的值是 tranStateTable.getMaxOffset()/ TSStoreUnitSize

![SouthEast 7][]

3. 事物回查

![复制代码][copycode.gif]

定时回查线程会定时扫描（默认每分钟）每个存储事务状态的表格文件，遍历存储事务状态的表格记录
    
    如果是已经提交或者回滚的消息调过过，
    
    如果是prepared状态的如果消息小于事务回查至少间隔时间（默认是一分钟）跳出终止遍历
    
    调transactionCheckExecuter.gotocheck方法向producer回查事物状态，
    
             根据group随机选择一台producer
    
             查询消息，根据commitLogOffset和msgSize到commitlog查找消息
    
             向Producder发起请求,请求code类型为CHECK_TRANSACTION_STATE，producer的DefaultMQProducerImpl.checkTransactionState()方法
             来处理broker定时回调的请求，这里构建一个Runnable任务异步执行producer注册的回调接口，处理回调，在调endTransactionOneway向broker
             发送请求更新事物消息的最终状态
    
    无Prepared消息，且遍历完，则终止扫描这个文件的定时任务

![复制代码][copycode.gif]

4. 事物消息的load&recover

TransactionStateService.load ()事物状态服务加载， 加载只是建立文件映射
     redoLog队列恢复，加载本地redoLog文件
     tranStateTable事物状态表， 加载本地tranStateTable文件

recover:

![复制代码][copycode.gif]

正常恢复：
    
             利用tranRedoLog文件的recover
    
             利用tranStateTable文件重建事物状态表
    
    异常恢复：
    
             先按照正常流程恢复TranRedo Log
    
             commitLog异常恢复，commitLog根据checkpoint时间点重新生成 redolog，重新分发消息DispatchRequest,
    
    分发消息到位置信息到ConsumeQueue
    
                      更新Transaction State Table
    
                       记录TransactionRedo Log
    
             删除事物状态表tranStateTable
    
    通过RedoLog全量恢复StateTable
    
             重头扫描RedoLog， 过滤出所有prepared状态的消息， 将commit或者rollback的消息对应的prepared消息删除
    
             重建StateTable,  将上面过滤出的prepared消息，添加到事物状态表文件中

![复制代码][copycode.gif]

这个事物状态表transstable的作用是定期（1分钟）将状态为prepared事物回查producer端redolog这个队列其实标记消费到哪了， 
    事物状态的恢复根本上是有commitlog来做的

consumer主要2中模式

pull  and push

标签:  [MQ+NIO][MQ_NIO]

好文要顶  关注我  收藏该文  ![icon_weibo_24.png][]  ![wechat.png][]

[![20151208093634.png][]][20151208093634.png 1]

[Givefine][20151208093634.png 1]  
[关注 - 0][- 0]  
[粉丝 - 133][- 133]

\+加关注

[« ][Link 1]上一篇： [分布式开放消息系统(RocketMQ)的原理与实践][Link 1]  
[» ][Link 2]下一篇： [大型网站技术架构（1）][Link 2]

posted @  2016-11-08 10:19  [Givefine][] 阅读( 2175) 评论( 0)  [编辑][Link 3]  [收藏][Link 4]

[arch]: https://github.com/a2888409/RocketMQ-Learning/raw/master/book/ch1/arch.jpg
[arch 1]: https://github.com/a2888409/RocketMQ-Learning/blob/master/book/ch1/arch.jpg
[SouthEast]: /images/20220608/fc0cd943141940b490c1393c49ef9c5b.png
[SouthEast 1]: /images/20220608/875b0d72351c408785088314caeee286.png
[copycode.gif]: /images/20220608/a2a608df8ef34cd5b9937ddff2732740.png
[SouthEast 2]: /images/20220608/e5b296ff842c4912ae128c8caa9d9c35.png
[SouthEast 3]: /images/20220608/75cd66acd7604554802841449de2dc53.png
[SouthEast 4]: /images/20220608/a24a72d4361940928d1a8783c4a3a0ba.png
[SouthEast 5]: /images/20220608/1ae0d01d0752436a80b8bca47c908e16.png
[SouthEast 6]: /images/20220608/b8f0a07f8f9f4ba9a615999f50aefcc3.png
[SouthEast 7]: /images/20220608/b317046e2b324931b6eafa3a12f4ee2b.png
[MQ_NIO]: http://www.cnblogs.com/wxd0108/tag/MQ%2BNIO/
[icon_weibo_24.png]: http://common.cnblogs.com/images/icon_weibo_24.png
[wechat.png]: /images/20220608/67854fba0d5347f68a8d9e356243907d.png
[20151208093634.png]: /images/20220608/9a6fbf3650ab4eb9a3fc792b3e23f748.png
[20151208093634.png 1]: http://home.cnblogs.com/u/wxd0108/
[- 0]: http://home.cnblogs.com/u/wxd0108/followees
[- 133]: http://home.cnblogs.com/u/wxd0108/followers
[Link 1]: http://www.cnblogs.com/wxd0108/p/6038543.html
[Link 2]: http://www.cnblogs.com/wxd0108/p/6046315.html
[Givefine]: http://www.cnblogs.com/wxd0108/
[Link 3]: https://i.cnblogs.com/EditPosts.aspx?postid=6041829
[Link 4]: http://www.cnblogs.com/wxd0108/p/6041829.html#