一张图进阶 RocketMQ - 整体架构

前言

一张图进阶 RocketMQ - 整体架构-LMLPHP

视频在 B 站同步更新（B站：三此君），欢迎围观一张图进阶 ROcketMQ -整体架构 (视频版)，记得调整视频清晰度哦~

本文是《一张图进阶 RocketMQ》系列的第 1 篇，今天的内容主要分为两个部分：

整体架构：会从大家熟悉的“生产者-消费者模式”逐步推出 RocketMQ 完整架构，只需要记住一张完整的架构图即可。
消息收发示例：通过 Docker 部署 RocketMQ，并用简单的示例串起 RocketMQ 消息收发流程。

整体架构

什么是消息队列？顾名思义，首先得有一个队列，这个队列用来存储消息。那有了消息队列就得有人往里面放，有人往里面取。有没有似曾相识的感 jio，这莫非就是连小学生都知道的，经典的“生产者-消费者模式”？接下来我们就来看看它里面穿了什么？

一张图进阶 RocketMQ - 整体架构-LMLPHP

别急，先来回顾一下 “生产者-消费者模式” 这个老朋友。简单来说，这个模型是由两类线程和一个队列构成：

生产者线程：生产产品，并把产品放到队列里。
消费者线程：从队列里面获取产品，并消费。

一张图进阶 RocketMQ - 整体架构-LMLPHP

有了这个队列，生产者就只需要关注生产，而不用管消费者的消费行为，更不用等待消费者线程执行完；消费者也只管消费，不用管生产者是怎么生产的，更不用等着生产者生产。

这意味着什么呢，生产者和消费者之间实现解藕和异步。这就老厉害了，你未来工作学习中会不断遇到这些概念，但是越看越看不懂。我们生活中很多都是异步的。比如最近新冠疫情卷土重来，我点的外卖只能送到小区门口的外卖队列里面，而我只能去外卖队列里面取外卖，然后一顿狼吞虎咽。

具体 “生产者-消费者模式” 怎么实现，想必各位小学都学过了，我们来看看这个模式还有什么问题吧。最大的问题就是我们小学学的 “生产者-消费者模式” 是个单机版的，只能自嗨。这就相当于，我就是外卖骑手，我点了个外卖放到外卖队列，然后我再从外卖队列里面去取，一顿操作猛如虎呀！于是就有了进化版，我们把消费者，队列，生产者放到不同的服务器上，这就是传说中的分布式消息队列了。

一张图进阶 RocketMQ - 整体架构-LMLPHP

生产者生产的消息通过网络传递给队列存储，消费者通过网络从队列获取消息。但是还有问题，消息可能有很多种，全都放在一起岂不是乱套了？我点的外卖和快递全都放在一起，太难找了吧。于是我们就需要区分不同类型消息，相同类型的消息称为一个 Topic。同时，骑手不可能只有一个，点外卖的也不会只有我一个人，于是就有了生产者组和消费者组。

一张图进阶 RocketMQ - 整体架构-LMLPHP

但还是有问题呀，小区那么大，一个队列放不下。我住在小区南门，点个外卖还要跑去北门拿，那真的是 eggs hurt。于是物业在东南西北门各设了一个外卖快递放置点。也就是我们有多个队列，组成 队列集群。

一张图进阶 RocketMQ - 整体架构-LMLPHP

可是，问题又双叒叕来了（还有完没完），一个小区那么多个外卖快递队列，骑手怎么知道送到哪里去，我又怎么知道去哪里取？很简单，导航呀。我们把导航的信息称为路由信息，这些信息需要有一个管理的地方，它告诉生产者，某这个 Topic 的消息可以发给哪些队列，同时告诉消费者你需要的消息可以从哪些队列里面取。RocketMQ 为这些路由信息的设置了管理员 NameServer，当然 NameServer 也可以有很多个，组成 NameServer 集群。

到这里，你就应该知道 RocketMQ 里面都穿了什么啦。包括了生产者（Producer），消费者（Consumer），NameServer 以及队列本身（Broker）。Broker 是代理的意思，负责队列的存取等操作，我们可以把 Broker 理解为队列本身。

一张图进阶 RocketMQ - 整体架构-LMLPHP

NameServer：我们可以同时部署很多台 NameServer 服务器，并且这些服务器是无状态的，节点之间无任何信息同步。
NameServer 起来后监听端口，等待 Broker、Producer、Consumer 连上来，NameServer 是集群元数据管理中心。
Broker：Broker 启动，跟所有的 NameServer 保持长连接，每 30s 发送一次发送心跳包（像心跳一样持续稳定的发送请求）。心跳包中包含当前 Broker 信息 ( IP+ 端口等）以及存储所有 Topic 信息。注册成功后，NameServer 集群中就有 Topic 跟 Broker 的映射关系。
Topic：收发消息前，先创建 Topic，创建 Topic 时需要指定该 Topic 要存储在哪些 Broker 上，也可以在发送消息时自动创建 Topic。
Producer：Producer 发送消息，启动时先跟 NameServer 集群中的其中一台建立长连接，并从 NameServer 中获取当前发送的 Topic 存在哪些 Broker 上，采用轮询策略从选择一个队列，然后与队列所在的 Broker 建立长连接，并向 Broker 发消息。
Consumer：Consumer 跟 Producer 类似，跟其中一台 NameServer 建立长连接，获取当前订阅 Topic 存在哪些 Broker 上，然后直接跟 Broker 建立连接通道，开始消费消息。

消息收发示例

RocketMQ 部署

刚刚我们了解 RocketMQ 整体架构，那怎么样通过 RocketMQ 收发消息呢？需要先通过 Docker 部署一套 RocketMQ：

如果你没有安装 Docker，可以根据菜鸟教程 MacOS Docker 安装/Windows Docker 安装进行安装。然后，通过 docker-compose 部署 RocketMQ：

克隆 docker-middleware 仓库，打开 dockers 目录；
修改./conf/broker.conf中的brokerIP1 参数为本机 IP；
进入docker-compose.yml文件所在路径，执行docker-compose up命令即可；

一张图进阶 RocketMQ - 整体架构-LMLPHP

部署完成后我们就可以在 Docker Dashboard 中看到 RocketMQ 相关容器，包括 Broker、NameServer 及 Console（RocketMQ 控制台），到这里我们就可以使用部署的 RocketMQ 收发消息了。

RocketMQ 已经部署好了，接下来先来看一个简单的消息收发示例，可以说是 RocketMQ 的 "Hello World"。

消息发送

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        // 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
        // 设置NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动Producer实例
        producer.start();
        // 创建消息，并指定Topic，Tag和消息体
        Message msg = new Message("Topic1","Tag", "Key",
                                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
        // 发送消息到一个Broker
        SendResult sendResult = producer.send(msg);
      	// 通过sendResult返回消息是否成功送达
        System.out.printf("%s%n", sendResult);
        // 如果不再发送消息，关闭Producer实例。
        producer.shutdown();
    }
}

首先，实例化一个生产者 producer，并告诉它 NameServer 的地址，这样生产者才能从 NameServer 获取路由信息。
然后 producer 得做一些初始化（这是很关键的步骤），它要和 NameServer 通信，要先建立通信连接等。
producer 已经准备好了，那得准备好要发的内容，把 "Hello World" 发送到 Topic1。
内容准备好，那 producer 就可以把消息发送出去了。producer 怎么知道 Broker 地址呢？他就会去 NameServer 获取路由信息，得到 Broker 的地址是 localhost:10909，然后通过网络通信将消息发送给 Broker。
生产者发送的消息通过网络传输给 Broker，Broker 需要对消息按照一定的结构进行存储。存储完成之后，把存储结果告知生产者。

消息接收

public class Consumer {

	public static void main(String[] args) throws InterruptedException, MQClientException {
    	// 实例化消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
    	// 设置NameServer的地址
        consumer.setNamesrvAddr("localhost:9876");
    	// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        erbconsumerijun.subscribe("sancijun", "*");
    	// 注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(
              List<MessageExt> msgs,ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                // 标记该消息已经被成功消费
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
	}
}

首先，实例化一个消费者 consumer，告诉它 NameServer 的地址，这样消费者才能从 NameServer 获取路由信息。
然后这个消费者需要知道自己可以消费哪些 Topic 的消息，也就是每个消费者需要订阅一个或多个 Topic。
消费者也需要做一些初始化，业务本身并没有理会怎么从 Broker 拉取消息，这些都是消费者默默无闻的奉献。所以，我们需要启动消费者，消费者会从 NameServer 拉取路由信息，并不断从 Broker 拉取消息。拉取回来的消息提供给业务定义的 MessageListener。
消息拉取回来后，消费者需要怎么处理呢？每个消费者都不一样（业务本身决定），由我们业务定义的 MessageListener 处理。处理完之后，消费者也需要确认收货，就是告诉 Broker 消费成功了。

以上就是本文的全部内容，本文没有堆砌太多无意义的概念，没有讲什么削峰解耦，异步通信。这些内容网上也很多，看了和没看没什么两样。

最后，看懂的点赞，没看懂的收藏。来都来了，交个朋友，遇到什么问题都可以和三此君分享呀。关注微信公众号：三此君。回复 mq，可以领取 RocketMQ 相关的所有资料。

一张图进阶 RocketMQ - 整体架构-LMLPHP

参考文献

RocketMQ 官方文档
RocketMQ 源码
丁威, 周继锋. RocketMQ技术内幕：RocketMQ架构设计与实现原理. 机械工业出版社, 2019-01.
李伟. RocketMQ分布式消息中间件：核心原理与最佳实践. 电子工业出版社, 2020-08.
杨开元. RocketMQ实战与原理解析. 机械工业出版社, 2018-06.

转载请注明出处

整体架构