RocketMQ源码-搭建源码调试环境

╰半橙微兮° 2023-09-23 22:21 143阅读 0赞

## 1. 基本架构 ##

RocketMQ架构上主要分为四部分，如下图所示:

![format_png][]

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

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

*  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的路由的信息。

*  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的消息进行索引服务，以提供消息的快速查询。

![format_png 1][]

## 2. 获取源码 ##

rocketMq项目的github仓库为[github.com/apache/rock…][github.com_apache_rock]。 我们选择4.8版本。

![Image 1][]

## 3. 本地启动 ##

拿到代码后，我们就开始进行本地启动了，没错，就是在idea中进行启动。

### 3.1 复制conf目录 ###

在启动项目前，我们需要进行一些配置，rocketMq项目的配置文件位于rocketmq/distribution模块下的conf目录中，直接整个复制到某个目录下（作为rocketmq的根目录，存放conf,log，store等内容）：

![format_png 2][]

也不需要改动，复制出来就行了，这些配置的内容后面分析源码时再讲解吧。

### 3.2 启动nameServer ###

nameServer的主类为org.apache.rocketmq.namesrv.NamesrvStartup：

![format_png 3][]

如果我们直接运行main()方法，会报错：

![format_png 4][]

报错信息已经很明确了，需要我们配置ROCKETMQ\_HOME目录，我们在idea中进行配置即可：

打开配置界面，填写ROCKETMQ\_HOME配置：

![format_png 5][]

这里我填写的是ROCKETMQ\_HOME=/Users/haoshaofei/IdeaProjects/rocketmq-home，这个ROCKETMQ\_HOME路径就是conf文件夹所在的目录。

填写好后，就可以启动了：

![format_png 6][]

### 3.3 启动broker ###

broker的主类为org.apache.rocketmq.broker.BrokerStartup，启动方式与nameServer很相似，启动前也要配置ROCKETMQ\_HOME路径：

![format_png 7][]

相比于nameServer，这里多需要指定配置文件conf:

#所属集群名字brokerClusterName=rocketmq-cluster
    #broker名字，注意此处不同的配置文件填写的不一样brokerName=broker-a
    #0 表示 Master，>0 表示 SlavebrokerId=0#nameServer地址，分号分割namesrvAddr=127.0.0.1:9876#在发送消息时，自动创建服务器不存在的topic，默认创建的队列数defaultTopicQueueNums=4#是否允许 Broker 自动创建Topic，建议线下开启，线上关闭autoCreateTopicEnable=true#是否允许 Broker 自动创建订阅组，建议线下开启，线上关闭autoCreateSubscriptionGroup=true#Broker 对外服务的监听端口listenPort=10911#删除文件时间点，默认凌晨 4点deleteWhen=04#文件保留时间，默认 48 小时fileReservedTime=120#commitLog每个文件的大小默认1GmapedFileSizeCommitLog=1073741824#ConsumeQueue每个文件默认存30W条，根据业务情况调整mapedFileSizeConsumeQueue=300000#destroyMapedFileIntervalForcibly=120000#redeleteHangedFileInterval=120000#检测物理文件磁盘空间diskMaxUsedSpaceRatio=88#存储路径storePathRootDir=/Users/haoshaofei/IdeaProjects/rocketmq-home/store
    #commitLog 存储路径storePathCommitLog=/Users/haoshaofei/IdeaProjects/rocketmq-home/store/commitlog
    #消费队列存储路径存储路径storePathConsumeQueue=/Users/haoshaofei/IdeaProjects/rocketmq-home/store/consumequeue
    #消息索引存储路径storePathIndex=/Users/haoshaofei/IdeaProjects/rocketmq-home/store/index
    #checkpoint 文件存储路径storeCheckpoint=/Users/haoshaofei/IdeaProjects/rocketmq-home/checkpoint
    #abort 文件存储路径abortFile=/Users/haoshaofei/IdeaProjects/rocketmq-home/abort
    #限制的消息大小maxMessageSize=65536#flushCommitLogLeastPages=4#flushConsumeQueueLeastPages=2#flushCommitLogThoroughInterval=10000#flushConsumeQueueThoroughInterval=60000#Broker 的角色#- ASYNC_MASTER 异步复制Master#- SYNC_MASTER 同步双写Master#- SLAVEbrokerRole=ASYNC_MASTER
    
    #刷盘方式#- ASYNC_FLUSH 异步刷盘#- SYNC_FLUSH 同步刷盘flushDiskType=ASYNC_FLUSH
    #checkTransactionMessageEnable=false#发消息线程池数量#sendMessageThreadPoolNums=128#拉消息线程池数量#pullMessageThreadPoolNums=128复制代码

添加启动参数：

![format_png 8][]

配置完成之后就可以启动了：

![format_png 9][]

### 3.4 启动管理后台 ###

rocketMq的管理后台在另一个仓库[github.com/apache/rock…][github.com_apache_rock 1]，除了后台，这个仓库还包含了许多的其他模块：

![format_png 10][]

我们并不需要分析这个项目，源码本可以不必下载，但我在找这个项目的release版本时，发现并没有提供已编译好的jar包，需要自己构建代码，因此我就再次下载了这个代码源码。

这个项目的代码我们并不分析，因此直接在master分支上操作即可，

管理后台项目为rocketmq-console，主类为org.apache.rocketmq.console.App：

![format_png 11][]

在启动前，我们需要修改下application.properties的配置，找到rocketmq.config.namesrvAddr配置，添加nameServer的ip与端口，这里我们连接的是本地应用，直接填写localhost:9876：

server.port=8180
    ...
    rocketmq.config.namesrvAddr=localhost:9876
    ...
    复制代码

启动，结果如下：

![format_png 12][]

访问[http://localhost:8180][http_localhost_8180]，结果如下：

![format_png 13][]

可以看到broker已经出现在cluster列表中了，这就表明启动成功了。

## 4. 收发消息测试 ##

rocketMq项目的example模块下有大量的测试示例，我们选择quickstart进行消息收发测试。

### 4.1 启动Consumer ###

我们先找到org.apache.rocketmq.example.quickstart.Consumer，增加：

![format_png 14][]

这个Consumer监听的topic是TopicTest，后面我们就会往这个topic发送消息。另外，需要注意nameServer的配置，我们是在本地启动的nameServer，因此这里配置的是127.0.0.1:9876。

运行main()方法，结果如下：

![format_png 15][]

### 4.2 启动Producer ###

我们找到 org.apache.rocketmq.example.quickstart.Producer 类，增加

![format_png 16][]

同样地，这里使用的是的nameServer地址是127.0.0.1:9876，topic 是TopicTest，运行，结果如下：

![format_png 17][]

再回过头看看Consumer的控制台：

![format_png 18][]

可以看到，Producer发送消息成功了，Consumer也成功获取到消息了。

### 4.3 异常分析 ###

如图所示：

![format_png 19][]

如果出现异常：

org.apache.rocketmq.client.exception.MQClientException: 
    No route info of this topic: TopicTest
    复制代码
    复制代码

这表明当前broker中没有TopicTest的topic，这时我们可以手动创建topic，也可以在启动时指定autoCreateTopicEnable=true.

如果是按上面步骤进行的，请确认下org.apache.rocketmq.broker.BrokerStartup是否配置启动参数

-n localhost:9876 autoCreateTopicEnable=true复制代码

配置方式就按3.3节的方式配置就行了。

## 5. 日志配置 ##

在启动的时候，日志配置文件是将日志打印到$\{user.name\}，自己认为不好找日志，就把日志打印到ROCKETMQ\_HOME下的logs配置文件中，需要将日志文件中的user.name替换为：ROCKETMQ\_HOME，然后重启，就会将日志记录到自定义的logs下:

![format_png 20][]

![format_png 21][]

## 6. 总结 ##

本文主要介绍了rocketMq的基本架构，通过源码展示了rocketMq的启动方式，最后通过rocketMq项目下example模块中的测试代码展示了消息的收发过程。

总的来说，本文还是在准备源码分析的环境，下篇文章开始，我们就正式开始rocketMq的源码分析了。

作者：hsfxuebao

原文链接：[https://juejin.cn/post/7188044776577433661][https_juejin.cn_post_7188044776577433661]

[format_png]: https://img-blog.csdnimg.cn/img_convert/3abf03588cbaf8278bde6000f6c18b56.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/01ef16514c241fc882c6ecf00792a81f.webp?x-oss-process=image/format,png
[github.com_apache_rock]: https://link.juejin.cn?target=https%3A%2F%2Fgithub.com%2Fapache%2Frocketmq%2Ftree%2Frocketmq-all-4.8.0
[Image 1]: 
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/6dd89ed8b9a37eb6478da2e968913f68.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/530b913c1657d74e3e21b3fed59eddce.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/1b9b18d157bff855e8f9fbc1c4a2e208.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/b1e408a8e924ec79e755122feed22fa1.webp?x-oss-process=image/format,png
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/1beb27d8453945f0c2892e0f4ebb53fb.webp?x-oss-process=image/format,png
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/67f1bdbbb8b1b589aac90788a41d36c5.webp?x-oss-process=image/format,png
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/5e04368150cb86bc9623fe10d7cc9018.webp?x-oss-process=image/format,png
[format_png 9]: https://img-blog.csdnimg.cn/img_convert/3791ae6d76391d1f56bb96dba2ecd941.webp?x-oss-process=image/format,png
[github.com_apache_rock 1]: https://link.juejin.cn?target=https%3A%2F%2Fgithub.com%2Fapache%2Frocketmq-externals
[format_png 10]: https://img-blog.csdnimg.cn/img_convert/451b0772487286e550be2d5fa30b9cb7.webp?x-oss-process=image/format,png
[format_png 11]: https://img-blog.csdnimg.cn/img_convert/200fe177b1a0010630aafd71e0de3e12.webp?x-oss-process=image/format,png
[format_png 12]: https://img-blog.csdnimg.cn/img_convert/f0c9d138766c34c07cf0ceb76000dc4e.webp?x-oss-process=image/format,png
[http_localhost_8180]: http://localhost:8180
[format_png 13]: https://img-blog.csdnimg.cn/img_convert/3789e82cf7cf41a549fc231c138f6ae0.webp?x-oss-process=image/format,png
[format_png 14]: https://img-blog.csdnimg.cn/img_convert/4faa03bc13c6ae79dcec2da52acdc5bd.webp?x-oss-process=image/format,png
[format_png 15]: https://img-blog.csdnimg.cn/img_convert/7a8664de152c2dc93148dccf6561b8b4.webp?x-oss-process=image/format,png
[format_png 16]: https://img-blog.csdnimg.cn/img_convert/dd43eff3f3dbf8b69657e2d9a41f3dcb.webp?x-oss-process=image/format,png
[format_png 17]: https://img-blog.csdnimg.cn/img_convert/5f7690b200e271b23cb97cd3d7149c57.webp?x-oss-process=image/format,png
[format_png 18]: https://img-blog.csdnimg.cn/img_convert/17e311c05a7dfe682c6b13bb557a7704.webp?x-oss-process=image/format,png
[format_png 19]: https://img-blog.csdnimg.cn/img_convert/1503a22edc6c7cf39bd77a50f919da03.webp?x-oss-process=image/format,png
[format_png 20]: https://img-blog.csdnimg.cn/img_convert/8c28a6413cd209b1f890e8f243562499.webp?x-oss-process=image/format,png
[format_png 21]: https://img-blog.csdnimg.cn/img_convert/a44343fec9a426ad309b82a9788a8446.webp?x-oss-process=image/format,png
[https_juejin.cn_post_7188044776577433661]: https://juejin.cn/post/7188044776577433661