IDEA 导入 RocketMQ 源码

前言

最近项目中有个功能需要在本地调试下 RocketMQ，所以需要在本地导入 RocketMQ 的源码并启动，故做此记录，便于回顾问题和与各位同学一起探讨。

一、RocketMQ 架构

在源码搭建前，需要先理解 RocketMQ 的四个重要组件，以及 RocketMQ 的工作流程：

在这里插入图片描述

NameServer 是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。
Broker 部署相对复杂，Broker 分为 Master 与 Slave，一个 Master 可以对应多个 Slave，但是一个 Slave 只能对应一个 Master，Master 与 Slave 的对应关系通过指定相同的 BrokerName，不同的 BrokerId 来定义，BrokerId 为 0 表示 Master，非 0 表示 Slave。Master 也可以部署多个。每个 Broker 与 NameServer 集群中的所有节点建立长连接，定时注册 Topic 信息到所有 NameServer。注意：当前 RocketMQ 版本在部署架构上支持一 Master 多 Slave，但只有 BrokerId=1 的从服务器才会参与消息的读负载。
Producer 与 NameServer 集群中的其中一个节点（随机选择）建立长连接，定期从 NameServer 获取 Topic 路由信息，并向提供 Topic 服务的 Master 建立长连接，且定时向 Master 发送心跳。Producer 完全无状态，可集群部署。
Consumer 与 NameServer 集群中的其中一个节点（随机选择）建立长连接，定期从 NameServer 获取 Topic 路由信息，并向提供 Topic 服务的 Master、Slave 建立长连接，且定时向 Master、Slave 发送心跳。Consumer 既可以从 Master 订阅消息，也可以从 Slave 订阅消息，消费者在向 Master 拉取消息时，Master 服务器会根据拉取偏移量与最大偏移量的距离（判断是否读老消息，产生读 I/O），以及从服务器是否可读等因素建议下一次是从 Master 还是 Slave 拉取。

结合部署架构图，描述集群工作流程：

启动 NameServer，NameServer 起来后监听端口，等待 Broker、Producer、Consumer 连上来，相当于一个路由控制中心。
Broker 启动，跟所有的 NameServer 保持长连接，定时发送心跳包。心跳包中包含当前 Broker 信息 ( IP+ 端口等) 以及存储所有 Topic 信息。注册成功后，NameServer 集群中就有 Topic 跟 Broker 的映射关系。
收发消息前，先创建 Topic，创建 Topic 时需要指定该 Topic 要存储在哪些 Broker 上，也可以在发送消息时自动创建 Topic。
Producer 发送消息，启动时先跟 NameServer 集群中的其中一台建立长连接，并从 NameServer 中获取当前发送的 Topic 存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
Consumer 跟 Producer 类似，跟其中一台 NameServer 建立长连接，获取当前订阅 Topic 存在哪些 Broker 上，然后直接跟 Broker 建立连接通道，开始消费消息。

二、环境准备

① JDK 1.8

java -version
② IntelliJ IDEA 2021
③ RocketMQ-4.9.x 源码
④ Maven
⑤ Git
⑥ Windows 11

三、下载源码

我们可以在 github 或者 gitee 上都能下载到 RocketMQ 的源码

github 上下载：

github 上搜素 rocketmq 就能找到：https://github.com/apache/rocketmq

在这里插入图片描述

gitee上下载：

gitee 上搜素 rocketmq 就能找到：https://gitee.com/apache/rocketmq

在这里插入图片描述

这里我是直接下载 ZIP 压缩包的，可以避免很多问题

在这里插入图片描述

解压：

在这里插入图片描述

四、编译源码

4.1 导入源码

打开 IDEA ，选择 File -> Open

在这里插入图片描述

选中 rocketmq 源码所在目录

在这里插入图片描述

导入进来是这样子的

在这里插入图片描述

项目导进来之后先检查下 JDK 的配置，配置 JDK 版本 1.8 的（建议 JDK 的版本不要配置太高）

在这里插入图片描述

配置你的 Maven

在这里插入图片描述

检查下 git 配置（编译的时候会自动去检测 git，所以需要检查下）

在这里插入图片描述

4.2 目录结构

在这里插入图片描述

4.3 运行程序

本地 Debug 环境搭建过程如下：

① 通过源码启动 Namesrv
② 通过源码启动 broker
③ 通过源码启动 Producer
④ 通过源码启动 Consumer

1. 启动 Namesrv

Namesrv 源码在 rocketmq-namesrv 包下，启动类是 src/main/java/org/apache/rocketmq/namesrv/NamesrvStartup.java，直接通过NamesrvStartup 的 main 方法启动会失败

在这里插入图片描述

终端输出提示我们需要配置一个 ROCKETMQ_HOME 环境变量，我们将环境变量配置到 IDEA 中

在这里插入图片描述

例如：ROCKETMQ_HOME=D:\source-code\rocketmq-4.9.x\rocketmq-4.9.x

在这里插入图片描述

再次启动 NamesrvStartup 后再次报错

在这里插入图片描述

意思是没有读到 conf 目录下的配置文件 logback_namesrv.xml

那就在项目下创建一个 conf 的文件夹

在这里插入图片描述

logback_namesrv.xml 这个配置文件可以在 distribution 的模块下找到，只需要将该文件复制一份到你所创建 conf 目录下即可

在这里插入图片描述

再启动 namesrv，控制台提示启动成功 The Name Server boot success.

在这里插入图片描述

到此 NameSrver 就算是启动完成了

2. 启动 Broker

Broker 源码在 rocketmq-broker 包下，启动类是 src/main/java/org/apache/rocketmq/broker/BrokerStartup.java，如果直接通过 IDEA 启动也是会失败

在这里插入图片描述

与上面 namesrv 一样，我们在 IDEA 上配置启动环境变量 ROCKETMQ_HOME

例如：ROCKETMQ_HOME=D:\source-code\rocketmq-4.9.x\rocketmq-4.9.x

在这里插入图片描述

再次启动还是会报与 Nameserv 相同的问题，我们只需要在 distribution 模块下找到 broker 相关的两个配置文件 logback_broker.xml 和 broker.conf 两个文件复制到 conf 目录下即可

在这里插入图片描述

再次启动，就可以看到 broker 也正常运行了

在这里插入图片描述

但是仔细观察会发现虽然 broker 启动成功了，但是 brokerName 好像和配置文件 broker.conf 中的 brokerName 不一致

在这里插入图片描述

我们可以启动 broker 时让它指定以 conf/broker.conf 的配置文件启动

在这里插入图片描述

再次启动

在这里插入图片描述

在 broker.conf 的配置文件中还有以下配置可以修改

# namesrv服务地址
namesrvAddr = 127.0.0.1:9876
# 运行自动创建topic，避免调试的时候麻烦
autoCreateTopicEnable = true
# 数据存储路径
storePathRootDir = D:/file/rocketmq/data_store
# commitlog存储文件
storePathCommitLog = D:/file/rocketmq/data_store/commitlog
# 消费队列存储文件
storePathConsumeQueue = D:/file/rocketmq/data_store/consumequeue
# 索引存储文件
storePathIndex = D:/file/rocketmq/data_store/index
# checkpoint存储文件
storeCheckpoint = D:/file/rocketmq/data_store/checkpoint
# abort文件
abortFile = D:/file/rocketmq/data_store/abort

做本地调试的时候最好添加以下两个配置：

# namesrv服务地址
namesrvAddr = 127.0.0.1:9876
# 运行自动创建topic，避免调试的时候麻烦
autoCreateTopicEnable = true

在这里插入图片描述

为了更好的查看启动 broker 的相关配置，可以在 logback_broker.xml 配置文件中的 RocketmqBroker 里面追加 <appender-ref ref="STDOUT"/> 配置，例如：

    <logger name="RocketmqBroker" additivity="false">
        <level value="INFO"/>
        <appender-ref ref="STDOUT"/>
        <appender-ref ref="RocketmqBrokerAppender"/>
    </logger>

在这里插入图片描述

3. 启动 Producer

在 example 模块中官方给了一个 producer 的示例

在这里插入图片描述

package org.apache.rocketmq.example.quickstart;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

public class Producer {

	// 发送消息次数：原本是 1000 次，作为演示我调整为 1 次
    public static final int MESSAGE_COUNT = 1;
    public static final String PRODUCER_GROUP = "please_rename_unique_group_name";
    public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";
    public static final String TOPIC = "TopicTest";
    public static final String TAG = "TagA";

    public static void main(String[] args) throws MQClientException, InterruptedException {

        DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP);
        // 下面这行代码原本是注释的，这里要放开注释
        producer.setNamesrvAddr(DEFAULT_NAMESRVADDR);

        producer.start();

        for (int i = 0; i < MESSAGE_COUNT; i++) {
            try {
                Message msg = new Message(TOPIC /* Topic */,
                    TAG /* Tag */,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
                );

                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
                Thread.sleep(1000);
            }
        }
        producer.shutdown();
    }
}

直接运行即可

在这里插入图片描述

可以看到是有发送一条消息的

4. 启动 Consumer

同样的在 example 模块中官方给了一个 consumer 的示例

在这里插入图片描述

package org.apache.rocketmq.example.quickstart;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;

public class Consumer {

    public static final String CONSUMER_GROUP = "please_rename_unique_group_name_4";
    public static final String DEFAULT_NAMESRVADDR = "127.0.0.1:9876";
    public static final String TOPIC = "TopicTest";

    public static void main(String[] args) throws InterruptedException, MQClientException {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMER_GROUP);
		// 下面这行代码原本是注释的，这里要放开注释
        consumer.setNamesrvAddr(DEFAULT_NAMESRVADDR);

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe(TOPIC, "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (msg, context) -> {
            System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msg);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();

        System.out.printf("Consumer Started.%n");
    }
}