【RocketMQ】源码以及环境搭建

1  前言

本节我们开始看一下 RocketMQ 相关的东西，我们主要看一条链路，大致如下：

（1）环境的搭建，源码的下载

（2）消息的结构以及相关类可能也会看下消息的存储

（3）消息的生产以及发送过程

（4）消息的消费过程

大概看着四方面的内容，本节主要看下源码的下载以及环境的搭建。

在看之前，我们顺便回顾一下 RocketMQ 有几个核心组件，要有个基本的认识：

（1）NameServer：是 RocketMQ 集群中的注册中心，它负责维护集群中所有生产者、消费者以及 Broker 的信息。它并不存储任何消息数据，而是维护着集群中各种组件之间的关系。

（2）Broker：Broker 是 RocketMQ 中实际存储消息的节点，它负责接收生产者发送的消息，并提供给消费者消费。Broker 可以组成集群，并且支持水平扩展，即可以通过增加 Broker 实例来提高系统的吞吐量。每个 Broker 可以包含多个 Topic，每个 Topic 又可以分为多个队列（Queue）。

（3）Producer：生产者是向 Broker 发送消息的应用程序或服务。生产者将消息发送到指定的 Topic 上，可以使用同步或异步的方式进行发送。

（4）Consumer：消费者是从 Broker 获取消息的应用程序或服务。消费者可以订阅一个或多个 Topic，并从 Broker 中拉取消息。RocketMQ 支持两种消费模式：Pull 模式（消费者主动从 Broker 拉取消息）和 Push 模式（Broker 将消息推送给消费者）。

（5）Topic：Topic 是消息的逻辑分类，生产者将消息发送到特定的 Topic，而消费者则订阅这些 Topic。每个 Topic 都由一个或多个队列组成，以支持消息的并行处理。

（6）Message Queue：Message Queue 是存储在 Broker 上的物理队列，它是消息存储的基本单位。一个 Topic 可以有多个 Message Queue，消费者通过均衡算法选择不同的队列进行消费。默认1个Topic会在1个broker上创建4个队列。

参考：源码以及可视化界面下载地址

另外关于 Linux 搭建 MQ 的话可以参考：【Centos】Centos 7.6 安装 RocketMQ 5.1

推荐书籍：《RocketMQ技术内幕》，一门技术必看一本书

2  环境准备

2.1  源码下载

上边的地址直接打开，下载如下：

下载完，然后下载依赖，最后效果如下：

我们这里顺便看一下核心目录都是干什么的：

（1）broker: broker 模块（broker 启动进程）。

（2）client ：消息客户端，包含消息生产者、消息消费者相关类。

（3）common：公共包。

（4）dev ：开发者信息（非源代码）。

（5）distribution ：部署实例文件夹（非源代码）。

（6）example: RocketMQ 示例代码。

（7）filter ：消息过滤相关基础类。

（8）filtersrv ： 消息过滤服务器实现相关类（Filter 启动进程）。

（9）logappender：日志实现相关类。

（10）namesrv: N ameServer 实现相关类（Names巳rver 启动进程）。

（11）openmessaging ： 消息开放标准，正在制定中。

（12）emoting ： 远程通信模块，基于 Netty。

（13）srvutil ：服务器工具类。

（14）store ：消息存储实现相关类。

（15）style: checkstyle 相关实现。

（16）test ： 测试相关类。

（17）tools ： 工具类，监控命令相关实现类。

2.2 本地 MQ 搭建

我们先下载一下二进制包，如下：

解压后如下：

然后我们开始启动（当然你本地要有 Java环境哈）：

（1）首先启动 namesrv:

start mqnamesrv.cmd

启动如果环境变量未配置的话如下：

可以添加一下环境变量如下：

启动 namesrv 的效果如下：

!!!窗口不要关噢。

（2）启动 broker:

start mqbroker.cmd -n 0.0.0.0:9876 

到这里，MQ算是启动完了，我们再看下可视化界面。

2.3  可视化界面

首先下载可视化源码包如下：

解压完编译后如下：

启动后效果：

2.3  环境验证

我们这里发送一条消息，试一下：

public class RocketMQProducerTest {

    @SneakyThrows
    public static void main(String[] args) {
        TransactionMQProducer producer = new TransactionMQProducer("TopicTest");
        producer.setNamesrvAddr("0.0.0.0:9876");
        producer.start();

        // 创建一个消息实例
        Message msg = new Message(
                "TopicTest", // topic
                "TagA",      // tag
                "OrderID18", // key
                "Hello RocketMQ".getBytes(RemotingHelper.DEFAULT_CHARSET)); // body

        // 发送消息
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);

        producer.shutdown();

    }

}

效果如下：

 行，没啥问题。

3  小结

本节我们主要准备下源码环境，简单体验一下 MQ 的启动过程，下节我们继续，有理解不对的地方欢迎指正哈。