RocketMQ Producer启动过程

Posted 2022-01-30 乐观男孩

目录

• 说明
• 启动过程
• 总结

说明

RocketMQ Producer启动过程，其实比较简单，主要是对涉及的组件初始化、启动定时任务、初始化Netty客户端等。先来看一下Producer启动过程中涉及到的类的关系

从类图中大概可以看出，一个Producer启动过程涉及哪几个类以及每个类大概的功能。从图中也可以看出，最终与服务端进行交互的，其实是NettyRemotingClient类。基于NettyRemotingClient基础之上，封装了一个对外提供操作的API操作类MQClientAPIImpl，该类大概的功能如下：
1、关于Topic的创建、删除、查询等操作；
2、关于Group的创建、删除、查询等操作；
3、关于Broker的更新配置、查询信息等操作
4、关于NameService的更新配置、查询配置等操作；
4、发送、拉取、查询、查看Message、Offset等跟Message的操作
5、查询、更新等Consumer相关操作
6、客户端（生产者/消费者）注册、注销、发送心跳

启动过程

1、DefaultMQProducer启动
该类的start()方法就是Producer启动的入口。其主要干的工作有两点：
（1）、启动defaultMQProducerImpl
（2）、如果启动了消息跟踪机制（构造函数通过enableMsgTrace指定），则启动TraceDispatcher。

2、DefaultMQProducerImpl启动
（1）、初始化MQClientInstance，将DefaultMQProducerImpl注册到MQClientInstance，然后启动MQClientInstance。

（2）、向所有Broker发送心跳信息 + 上传Consumer的过滤规则（针对消费者有用，只有是Push消费模式才上传）
（3）、启动定时任务，清除过期的请求（已发送请求，但是没有响应）

3、MQClientInstance启动
（1）、启动MQClientAPIImpl：通过启动MQClientAPIImpl来启动NettyRemotingClient，NettyRemotingClient会初始化与Netty相关的组件，Netty与服务器进行交互。
（2）、启动好几个定时任务：
a、请求服务端获取最新的NameServer地址；延时10秒启动，间隔两分钟执行一次
b、从NameServer获取Topic路由信息并更新到本地；延时10毫秒启动，默认间隔30秒执行一次
c、清除已掉线的Broker信息，根据定时任务b获取的最新数据进行判断，如果Broker还在最新获取到的数据里，证明还在线，否则不在线；向所有的Broker发送心跳信息；延时1秒启动，默认间隔30秒执行一次
d、保存所有Consumer的Offset，如果是广播模式，保存到本地文件，如果是集群模式，保存到远程Broker；延时10秒启动，间隔5秒执行一次
e、调整Consumer的线程池数量，只有是PushConsumer才调整。
（3）、启动PullMessageService：拉取消息服务，主要在实例为消费者时才有用，内部执行时，从Broker中拉取消息；
（4）、启动RebalanceService：对消费者消费Topic哪些队列进行重新均衡
（5）、启动内部DefaultMQProducer：DefaultMQProducer->DefaultMQProducerImpl->MQClientInstance是一层关系，然后在MQClientInstance内再创建DefaultMQProducer->DefaultMQProducerImpl->MQClientInstance。

总结

1、RocketMQ Producer底层使用Netty进行通信。
2、封装统一的接口API对外提供服务。
3、启动流程是一个链条形式，启动一个入口组件，在入口组件内部对依赖的组件进行启动，一级级传下去。这种方式在很多框架中都使用（比如Tomcat）。