RocketMQ：深入理解Broker如何接收Producer生产消息请求？

清疚 2023-10-02 13:50 3阅读 0赞

## 一、前言 ##

今天原打算写一篇关于消息在Broker端如何存储的文档，刚梳理好整体文章脉络，想到一个问题：Broker端是如何接收生产消息请求的？印象中是通过Netty接收的，由于对具体的实现比较好奇，也就有了今天的故事。

## 二、源码分析 ##

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16]

以下所有分析相关的源码注释请见GitHub中的release-4.8.0分支：[https://github.com/Saint9768/rocketmq/tree/rocketmq-all-4.8.0][https_github.com_Saint9768_rocketmq_tree_rocketmq-all-4.8.0]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_13_color_FFFFFF_t_70_g_se_x_16]

### 1、RemotingServer消息生产/消费服务启动 ###

又要开始快乐的找入口环节了，我们先看看Broker作为服务端对外提供的服务RomotingServer。

> RemotingServer是一个Netty服务端服务，它开始了10911端口用来处理Producer和Consumer的网络请求。

在前面我们介绍过RocketMQ中Broker的启动流程：[RocketMQ：深度剖析Broker启动流程原理、源码][RocketMQ_Broker]。remotingServer就是在BrokerController\#start()方法中启动的。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
remotingServer是一个接口，点进去我们发现其有两个实现，分别是：`NettyRemotingServer`、`NettyRemotingClient`。  
![在这里插入图片描述][035695547e8e497bab8da4a48d5b01d0.png]  
对于Producer和Consumer客户端而言，broker是作为服务端的存在，所以我们要看的是`NettyRemotingServer#start()`方法。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 2]  
emmmm，这是一个标准的`netty服务启动流程`。对于netty相关的内容，我们这里仅需要关注`pipeline`上的channelHandler（`处理读写请求的操作`），在`NettyRemotingServer`中处理读写请求的`ChannelHandler`为serverHandler(`NettyServerHandler`)；代码如下图所示：  
![在这里插入图片描述][a0a899372db94feaac2f72d1c836323b.png]  
`NettyServerHandler`是`NettyRemotingServer`的一个内部类；里面会调用当前外部类对象`NettyRemotingServer`父类`NettyRemotingAbstract`的`processMessageReceived()`方法。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 3]  
这里会判断`RemotingCommand`命令的类型是请求还是响应。对于消息生产而言，RemotingCommand的命令类型为请求，所以我们接着看一下`processRequestCommand()`方法。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 4]  
首先根据命令的code从processorTable中获取相应的事件处理器和线程池，如果获取不到则采用默认的事件处理器（`AdminBrokerProcessor`）；紧接着调用事件处理器处理请求，即核心处理请求的核心逻辑在`processor#asyncProcessRequest()`方法中。  
![在这里插入图片描述][7ff1d6ae80244754a373c0f73f891349.png]  
点进去我们发现`asyncProcessRequest#asyncProcessRequest`是一个抽象类的方法；对应生产者发送消息的处理器是`SendMessageProcessor`。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 5]  
**问题又来了？这个SendMessageProcessor是什么时候注册到processorTable中的？**

### 2、Processor事件处理器注册 ###

上面我们提到了一个问题：SendMessageProcessor是什么时候注册到processorTable中的？  
`processorTable`是`NettyRemotingAbstract`的成员变量，它里面的内容是在`BrokerController`初始化时注册的。BrokerController执行初始化的方法BrokerController\#initialize()方法中或调用registerProcessor()方法，其会将一大堆的Processor注册到processorTable中，比如：pullMessageProcessor、SendMessageProcessor、ConsumeManageProcessor等。  
![在这里插入图片描述][1baff483fb644680b83f1df97f6776a6.png]  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 6]  
然后我们发现并没有直接操作`processorTable`的地方，看这个`remotingServer#registerProcessor()`方法，注册处理器，感觉就是它了，我们再点进去看看：  
![在这里插入图片描述][5e05a9a2eb81430e868e37707a864d2c.png]  
其是个抽象方法，和上面一样，此时的Broker是作为服务端的存在，我们依旧看一下`NettyRemotingServer`是怎么做的。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 7]  
在注册Processor时，会现获取相应的线程池，如果线程池为null，则使用公用的线程池；这里是一个`线程池隔离`的机制。最后把Processor放到processorTable，这个注册事件处理器的流程就完事了。

### 3、接收Producer生产消息请求 ###

`Producer`发送消息到broker时，发送请求的code为`SEND_MESSAGE`。结合我们上面的分析：

*  当消息过来时，`NettyRemotingServer`会使用`NettyServerHandler`这个`ChannelHandler`来处理请求；
 *  接着会调用到`NettyRemotingAbstract`\#`processRequestCommand()`方法，根据请求命令的code获取对应的Processor来处理请求。
 *  从Processor的注册流程来看，处理`SEND_MESSAGE` 类型请求命令的Processor为`SendMessageProcessor`。

![在这里插入图片描述][44d50e74c1a8413584346585a0c104a8.png]  
没啥特别的，接着往下跟，进入到同名方法`asyncProcessRequest()`中。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 8]  
由于我们请求的Code不是`CONSUMER_SEND_MSG_BACK`，所以会进入到Switch代码块的default中；接着我们进入到：`asyncSendMessage()`方法中，看一下生产单个消息是怎么做的？  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 9]  
这个方法是在准备要发送消息的数据，主要工作如下：

1.  如果没有指定queue，就随机选一个queue；一般情况下也不会指定queue，除非顺序生产消息时。
2.  构建`MessageExtBrokerInner`对象，给消息添加一些额外信息，比如：`queue`、`topic`、`事务消息标识`、发送地址、发送时间等。
3.  根据是否是`事务消息`，决定发送事务消息还是普通消息。

后面MessageStore登场，进行消息的持久化操作，这一部分内容，我们放在下一篇文章中探讨。

## 三、原理总结 ##

**Broker端接收Producer生产消息的流程如下：**

*  在Broker启动时，即BrokerController\#start()方法中，初始化接收消息的Netty服务NettyRemotingServer。
 *  在`NettyRemotingServer`中`pipeline`下的`NettyServerHandler`这个`ChannelHandler`负责处理消息生产请求；
 *  接着调用到`NettyRemotingAbstract`的`processMessageReceived()`方法中根据请求的code调用对应的事件处理器处理请求。
 *  对于消息生产而言，最终会调用到`SendMessageProcessor#asyncProcessRequest()`方法处理消息生产请求。
 *  然后调用`SendMessageProcessor#asyncSendMessage()`方法封装消息、填充queue、topic信息，假如我们不指定queue进行消息生产，broker会`随机选择一个queue` 就是在这里做的；最后调用`MessageStore#asyncPutMessage()`方法持久化消息，最终将消息写到`CommitLog`中。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/f23ce34878fa4970aa6dc51687e3347d.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[https_github.com_Saint9768_rocketmq_tree_rocketmq-all-4.8.0]: https://github.com/Saint9768/rocketmq/tree/rocketmq-all-4.8.0
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_13_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/cecee6393cc140c6aa35b06f42ac523b.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_13,color_FFFFFF,t_70,g_se,x_16
[RocketMQ_Broker]: https://blog.csdn.net/Saintmm/article/details/120934879
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/2e24e893544a4705a6104b911f1a2df4.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[035695547e8e497bab8da4a48d5b01d0.png]: https://img-blog.csdnimg.cn/035695547e8e497bab8da4a48d5b01d0.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 2]: https://img-blog.csdnimg.cn/05ec2d497c6f4d1e86b81268dc89f0b9.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[a0a899372db94feaac2f72d1c836323b.png]: https://img-blog.csdnimg.cn/a0a899372db94feaac2f72d1c836323b.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 3]: https://img-blog.csdnimg.cn/53d18d446a8a48349f4018f99b32c939.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 4]: https://img-blog.csdnimg.cn/44b64dbfaa1f486fbcc0ac8463a83b6b.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[7ff1d6ae80244754a373c0f73f891349.png]: https://img-blog.csdnimg.cn/7ff1d6ae80244754a373c0f73f891349.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 5]: https://img-blog.csdnimg.cn/a01bbdd199924abba8e335c913aef79c.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[1baff483fb644680b83f1df97f6776a6.png]: https://img-blog.csdnimg.cn/1baff483fb644680b83f1df97f6776a6.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 6]: https://img-blog.csdnimg.cn/8d76e5962c964d59acac85211c468361.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[5e05a9a2eb81430e868e37707a864d2c.png]: https://img-blog.csdnimg.cn/5e05a9a2eb81430e868e37707a864d2c.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 7]: https://img-blog.csdnimg.cn/26184f849c9543ebb121fad579c6b1ea.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[44d50e74c1a8413584346585a0c104a8.png]: https://img-blog.csdnimg.cn/44d50e74c1a8413584346585a0c104a8.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 8]: https://img-blog.csdnimg.cn/869eea8bd77147cfa6cb0270fbd4a613.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5qKm5Zue546W5p-S_size_20_color_FFFFFF_t_70_g_se_x_16 9]: https://img-blog.csdnimg.cn/2a05e52633d74d25b3c086861679cfd1.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5qKm5Zue546W5p-S,size_20,color_FFFFFF,t_70,g_se,x_16