和你一起遇见更好的自己
后端与中间件 · 精读

RocketMQ系列-架构设计之启动

发布于 2021-09-28 系列：RocketMQ 系列 标签：RocketMQ / 消息队列 归档
本文写于 2021 年，涉及的组件版本、配置与数据可能已过时，请按经验材料阅读并核对当前版本。原文链接与内容保持不变。

之前我们已经学习了RocketMQ是由四个部分组成的。这篇文章更加深入的来看看这个四个部分。

NameServer

接下来，我们来看 NameServer 的相关内容，我们都知道 NameServer 是 RocketMQ 的注册中心。那它肯定会有 服务发现，检查检查，路由等等功能， 我们就按照这个思路去看看 NameServer 是如何启动的。

NameSrv 的启动过程

我们在部署 RocketMQ的时候，使用下面的命令启动了RocketMQ的NameServer.

nohup sh mqnamesrv &

这条命令其实执行的是：

sh ${ROCKETMQ_HOME}/bin/runserver.sh org.apache.rocketmq.namesrv.NamesrvStartup $@

也就说，这条命令执行运行的 NamesrvStartup 类。

NameSrv 的启动其实就两步：

  • 封装配置参数： 根据命令行参数封装 NameSrvConfig 配置。 可以执行配置文件，会解析配置文件中的相关配置。
  • 初始化并启动 NamesrvController.
    • 初始化:
      • 对 NameSrvController 进行配置，
      • 创建远程 netty Server,
      • 注册 DefaultRequestProcessor ，处理各种连接请求，
      • 创建了两个定时任务：每 10s 扫描一次 Broker 列表，移除不存活的 Broker 。每10s打印一次配置属性。
      • 创建配置文件监听器，监听配置文件是否有变化。
    • 启动：
      • 启动 netty Server : 实际上启动一个Netty服务。
      • 启动文件监听器。

在启动过程中, Broker 的健康检测是通过定时任务来实现的。那路由功能是怎么实现的呢？

这就要看 在 NameServer 中最重要的一个类了： DefaultRequetProcessor . DefaultRequestProcessor 封装了各种连接请求的处理。比如Broker的注册，根据Topic获取路由信息等等。具体可以参考如下代码:

@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {

    if (ctx != null) {
        log.debug("receive request, {} {} {}",
                request.getCode(),
                RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                request);
    }

    switch (request.getCode()) {
        case RequestCode.PUT_KV_CONFIG:
            return this.putKVConfig(ctx, request);
        case RequestCode.GET_KV_CONFIG:
            return this.getKVConfig(ctx, request);
        case RequestCode.DELETE_KV_CONFIG:
            return this.deleteKVConfig(ctx, request);
        case RequestCode.QUERY_DATA_VERSION:
            return queryBrokerTopicConfig(ctx, request);
        case RequestCode.REGISTER_BROKER:
            Version brokerVersion = MQVersion.value2Version(request.getVersion());
            if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
                return this.registerBrokerWithFilterServer(ctx, request);
            } else {
                return this.registerBroker(ctx, request);
            }
        case RequestCode.UNREGISTER_BROKER:
            return this.unregisterBroker(ctx, request);
        case RequestCode.GET_ROUTEINFO_BY_TOPIC:
            return this.getRouteInfoByTopic(ctx, request);
        case RequestCode.GET_BROKER_CLUSTER_INFO:
            return this.getBrokerClusterInfo(ctx, request);
        case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
            return this.wipeWritePermOfBroker(ctx, request);
        case RequestCode.GET_ALL_TOPIC_LIST_FROM_NAMESERVER:
            return getAllTopicListFromNameserver(ctx, request);
        case RequestCode.DELETE_TOPIC_IN_NAMESRV:
            return deleteTopicInNamesrv(ctx, request);
        case RequestCode.GET_KVLIST_BY_NAMESPACE:
            return this.getKVListByNamespace(ctx, request);
        case RequestCode.GET_TOPICS_BY_CLUSTER:
            return this.getTopicsByCluster(ctx, request);
        case RequestCode.GET_SYSTEM_TOPIC_LIST_FROM_NS:
            return this.getSystemTopicListFromNs(ctx, request);
        case RequestCode.GET_UNIT_TOPIC_LIST:
            return this.getUnitTopicList(ctx, request);
        case RequestCode.GET_HAS_UNIT_SUB_TOPIC_LIST:
            return this.getHasUnitSubTopicList(ctx, request);
        case RequestCode.GET_HAS_UNIT_SUB_UNUNIT_TOPIC_LIST:
            return this.getHasUnitSubUnUnitTopicList(ctx, request);
        case RequestCode.UPDATE_NAMESRV_CONFIG:
            return this.updateConfig(ctx, request);
        case RequestCode.GET_NAMESRV_CONFIG:
            return this.getConfig(ctx, request);
        default:
            break;
    }
    return null;
}

这段代码里实现了所有的Broker，Producer，Customer和NameServer交互的处理逻辑，包括路由功能等等。

NameServer的关闭

NameServer 的关闭就非常简单了.

NameSrvStartUp 在 initialize 和 start 之间，加入了 一个 关闭事件的监听器.

// 添加关闭的回调。
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
    @Override
    public Void call() throws Exception {
        controller.shutdown();
        return null;
    }
}));

内部是使用 NameSrvController的shutdown方法。 主要进行：

  • 关闭Netty服务
  • 关闭线程池
  • 关闭scheduledExecutorService
  • 关闭文件监听器

Broker

  • Broker的启动过程
  • Broker怎么样进行消息存储的
  • Broker的内部运行原理是什么样的？
  • 关闭流程

在 RocketMQ 的 Broker 这个启动环节下，我们可以直接找到 BrokerStartUp.java 这个类。

Broker的启动过程

Broker的启动本质上是启动了 一个Netty服务端和一个Netty客户端。 使用Netty客户端完成向NameSrv的注册，心跳检测，等数据交互。 使用Netty服务端处理 Producer发送的消息，并将消息按照不同的消息类型存储下来。

首先 Broker 依然还是会加载 Broker 的相关配置，包括：a.命令行中指定的参数,比如mqbroker,-c等， b.-c指定的文件中的Broker属性。然后根据加载的NettyServerConfig(启动的Netty服务端的配置),NettyClientConfig(启动的Netty客户端的配置), BrokerConfig(Broker的配置), messageStoreConfig(存储消息的配置)创建 BrokerController. 之后及时初始化BrokerController，注册Shutdown回调. 最后启动BrokerController。 shutdown回调其实没有什么可说的。我们再简单的看看初始化过程。

初始化过程

初始化过程是一个 "漫长" 的过程. 一开始是在去加载一些数据，初始化会去加载 创建的topic数据，消息的消费偏移量，广播组，消费过滤数据等的数据，然后创建 MessageStore 对象，作用是存储消息。注意这个 MessageStore是支持插件的形式扩展的。 如果数据加载成功之后，就会启动 netty 服务端。 代码中 启动了两个netty服务端: remotingServer 和 fastRemotingServer。然后就是初始化了一些线程池 用于注册Processor。 在注册 Processor的过程中，可以看到 fastRemotingServer 和 remotingServer 除了端口不一样之外， fastRemotingServer 没有 注册 PullMessageProcessor . 也就是说 fastRemotingServer 不支持 pullMessage 请求。然后 BrokerController 创建很多的定时任务，比如：定时记录每天的消息数据，定时持久化消费者消息. 定时持久化 消费者过滤时的数据情况 等。再就是 更新Broker的NameServer地址。 最后初始化事务，初始化权限，初始化RPC钩子。

启动过程

在完了初识话之后，下一步就是启动了。和NameServer一样的流程，先初始化，在start. 启动过程分为以下几步:

  • 启动 messageStore
  • 启动 remotingServer
  • 启动 fastRemotingServer
  • 启动文件监听器
  • 启动 BrokerOutAPI , 向 NameServer 服务端发送相关请求的连接与断开等，定时扫描ResponseTable并触发回调。
  • 启动PullRequestHoldService: 存储 pull Message 的请求, 并触发执行pull Message.
  • 启动定时任务，定时扫描不存活的生产者，消费者，消息过滤服务(非tag过滤)。
  • 启动消息过滤服务。 消息过滤服务并非是基于tag的消息的过滤，而是在Broker端提供了一种更加细粒度的消息过滤控制。
  • Broker的容灾处理
  • 定时任务: 注册 Broker 到 NameServer
  • 启动broker统计, 无动作
  • 清理过期请求。

关闭过程

这个关闭流程这里就不多说了，就是把上面启动的过程挨个关闭就好了。具体细节可以参考 org.apache.rocketmq.broker.BrokerController#shutdown() 我们后面也会分析这部分的代码。 除了关闭上面启动的服务之外，在关闭的时候，需要将消息进行持久化。 比如 ConsumerOffset, ConsumerFilter, 这也是当服务再次启动时保证消息能够正常被消费的保障。

Producer

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

我们从最简单的一个示例来看，生产者的启动流程

// 使用GroupName初始化Producer
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
// 指定NameSrv的地址: 也可以通过环境变量NAMESRV_ADDR来指定，则不需要下面这一行。
producer.setNamesrvAddr("name-server1-ip:9876;name-server2-ip:9876");
// 启动实例
producer.start();
try {
    // 创建消息实例,指定 topic, tag, message body.
    Message msg = new Message("TopicTest" /* Topic */,
        "TagA" /* Tag */,
        ("Hello RocketMQ !!").getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
    );
    // 发送消息给Broker
    SendResult sendResult = producer.send(msg);
    System.out.printf("%s%n", sendResult);
} catch (Exception e) {
    e.printStackTrace();
    Thread.sleep(1000);
}
// 关闭生产者
producer.shutdown();

关于 生产者角色，我们应该了解什么？或者说，看到上面的代码，你想更深入的知道些什么吗?(我听到了你说，不想..)

  • 生产者是如何启动? 启动过程中生产者都做了哪些事情?
  • 生产者是如何和Namesrv进行交互的?交互是什么信息?生产者是如何和NameSrv进行健康检查的?
  • 生产者是怎么样发送消息的?发送消息的过程是什么样的?
  • 生产者的关闭流程是什么样的?

生产者启动流程

封装生产者的属性

DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");

我们可以通过 DefaultMQProducer 类创建一个生产者对象。 这个类我们打算发送消息的程序入口。需要注意的是：这个类的实例是线程安全的：在配置并启动进程后，该类可以被视为线程安全的，可以在多个线程上下文中使用。

DefaultMQProducer 提供了5个构造参数(4.9.1版本, 并非在一个构造方式中)。

  • String namespace: 生产者实例的命名空间。可以理解为 MQ生产者的名称。
  • String producerGroup: 生产者组。 生产者组在概念上聚合了完全相同角色的生产者实例。 这对事务消息非常重要。对非事务性消息就没有太大关系了。(=>小白说:一类生产者。)
  • RPCHook rpcHook: 每个远程命令执行要执行的 RPC 回调。RPCHook 是一个接口, 提供了两个方法doBeforeRequest和doAfterResponse，表示在执行请求之前和接收返回之后分别执行相关逻辑；
  • boolean enableMsgTrace: 是否开启消息追踪
  • String customizedTraceTopic: 消息追踪日志使用的队列名字
  • String nameSrvAddr: 这个字段没有在构造方法中，我们可以手动调用set方法进行设置，也可以通过系统变量的形式进行设置。

生产者在设置完对应的参数之后，就会调用Start()方法。 start方法会设置group,然后调用在 构造方法中初始化的 defaultMQProducerImpl 实例。

如上图

  • 生产者首先会检查 producerGroup 的合法性。
  • 然后设置ProducerGroup
  • 创建 MQClientFactory. 将producer注册到MQClientInstance中，
  • 初始化topicPushlishInfo. topicPushlishInfo 主要用于存放消息的路由信息。
  • 然后通过 mQClientFactory.start() 完成启动，这一步骤很重要，我们来看一下里面的具体实现.
    • 1.首先会设置 NameServerAddr。前面有说过可以通过 DefaultProducer 的setNameSrvAddr方法手动设置，也可以通过系统变量的方式进行设置NameServer的地址：System.setProperty("rocketmq.namesrv.domain", "localhost");
    • 2.开启定时任务: 总共启动了5个定时器任务，分别是：定时更新NameServerAddr信息，定时更新topic的路由信息，定时清理下线的broker，定时持久化Consumer的Offset信息，定时调整线程池；
    • 3.pullMessageService和rebalanceService被用在消费端的两个服务类，分别是：从broker拉取消息的服务和均衡消息队列服务，负责分配消费者可消费的消息队列
  • 同步的发送心跳给所有的Broker.
  • 开启定时任务：定时扫描过期的请求。

生产者发送消息

生产者启动完成之后，我们再看一下发送消息的过程: 发送消息的逻辑主要是在 sendDefaultImpl 方法中。主要逻辑分成三步:1.获取队列的路由信息，2.获取 MessageQueue, 3.发送消息。

获取队列路由信息

在 启动生产者流程中，已经将topic的路由信息存储到了 topicPushlishInfo 中，并以 producerGroup 为 key, topicPushlishInfo 为Value ，存储到 topicPublishInfoTable 这个Map中。

获取路由信息则是通过 mQClientFactory.updateTopicRouteInfoFromNameServer(topic, true, this.defaultMQProducer) 获取的。此方法根据topic获取路由信息，具体连接哪台nameServer，会从列表中顺序的选择nameServer，实现负载均衡；

获取MessageQueue

获取 MessageQueue 则是通过 this.mqFaultStrategy.selectOneMessageQueue(tpInfo, lastBrokerName) 来实现的。 MQFaultStrategy 这个类实现了选择 MessageQueue 的策略。主要有四种策略:

  • latencyFaultTolerance：延迟容错对象，维护brokers的延迟信息；
  • sendLatencyFaultEnable：延迟容错开关，默认不开启；
  • latencyMax：延迟级别数组；
  • notAvailableDuration ：根据延迟级别，对应broker不可用的时长；

获取MessageQueue之后需要判定其对应的Broker是否可用，同时也需要和当前指定的brokerName进行匹配；如果没有获取到就选择一个延迟相对小的，pickOneAtLeast会做排序处理；如果都不行就直接获取一个MessageQueue，不管其他条件了

发送消息

首先需要获取指定broker的地址，这要才能创建channel与broker连接；然后就是一些hook处理；接下来就是准备发送的消息头SendMessageRequestHeader，最后根据不同的发送策略执行发送消息。

之前的文章中说过，RocketMQ发送消息有三种方式：同步，异步和单向。具体的使用方法可以参考这篇文章。在RocketMQ的生产者端可以发送多种类型的消息包括：延迟消息，顺序消息以及事务消息， 各种消息的发送过程我会在后面通过源码仔细和大家一起学习~。

生产者关闭流程

生产者的关闭主要有四步

  • 取消生产者的注册.
  • 关闭MqClient=> 主要是关闭生产者,关闭拉取消息服务，关闭定时任务服务，关闭远程client，关闭负载均衡服务这5种服务。
  • 关闭Producer的定时任务
  • 修改状态。

这样就完成了生产者的关闭流程。

生产者总结

生产者首先需要设置namesrv，或者指定其他方式更新namesrv；然后从namesrv获取topic的路由信息，路由信息包括broker以及Message Queue等信息，同时将路由信息保存在本地内存中，方便下次使用；最后从Message Queue列表中选择合适的Queue发送消息，实现负载均衡；

Consumer

现在我们还剩下 消费者 的启动流程了。从一个简单的例子说起:

从图中可以看到使用MQ的消费者主要分成三部分： 1.创建消费者对象, 2.配置消费的属性 nameServer,消费起点,订阅主题，回调事件等。3.启动消费者。

Consumer的启动过程

正如上文所说， 启动MQ消费者主要分为三部分，我们主要讲述第一部分和第三部分: 创建Consumer对象，启动。

创建 Consumer 对象

RocketMQ 支持两种消息消费模式, pull模式 和 push 模式。 pull 模式是消费者主动拉取消息, push 模式是 broker端主动推送消息给消息者端。可想而知，push模式是不管消费者端死活的，只要有消息就会推给消费者端，不管消费者是否能消费完。而pull模式是不管Broker端的，可能会造成消息积压的问题。

RocketMQ分别提供了pull模式 和 push 模式的消费者的支持。类结构如下图：

图中 push模式，提供了 MQPushConsumer(DefaultMQPushConsumer) 类来实现。 pull 模式提供 MQPullConsumer(DefaultMQPullConsumer) 类来实现。但是这个已经标记为废弃，并在2022年会移除，提供了 LitePullConsumer(DefaultLitePullConsumer) 来实现Pull模式。

创建 DefaultMQPushConsumer 对象

DefaultMQPullConsumer 对象的构造参数

  • consumerGroup: 消费者组
  • namespace: 生产者的 Namespace
  • allocateMessageQueueStrategy: 消息队列分配算法
  • rpcHook: rpc的钩子，用在远程调用之前执行
  • enableMsgTrace: 是否跟踪消息轨迹
  • customizedTraceTopic: 跟踪消息轨迹使用topic

在 DefaultMQPushConsumer 的构造方法中会创建 DefaultMQPushConsumerImpl 对象，我们后面所说的启动过程，其实就是 DefaultMQPushConsumerImpl 的启动过程，即 DefaultMQPushConsumerImpl.start().

创建 DefaultLitePullConsumer 对象

DefaultLitePullConsumer 对象的参数:

  • namespace: 生产者的命名空间
  • consumerGroup: 消费者组
  • rpcHook: RPC 的回调钩子

在 DefaultLitePullConsumer 的构造方法中会创建 defaultLitePullConsumerImpl 对象。而后面的启动过程，即是 DefaultLitePullConsumerImpl 的启动过程。即 DefaultLitePullConsumerImpl.start().

启动

这里我再把使用 DefaultMQPushConsumer 消费消息的案例 粘贴到这里

我们主要看 consumer.start() 内部的具体实现。

push模式: DefaultMQPushConsumerImpl.start()
  • 检查必要的参数consumerGroup,消费模式，消费起点，负载策略等。
  • 拷贝订阅关系，绑定到重试topic,以防止消费者 ack失败。
  • 创建MQClientInstance 实例。这里是一种单例模式。
  • 配置消费者再平衡的消费者组，消息模式，消息的分配策略， MQClientInstance 实例。
  • 实例化消息拉取的包装类 并注册消息过滤的钩子
  • 加载消息的消费偏移量。 如果是广播消息从本地获取偏移数据，如果是集群消息的话，则从远程获取偏移数据
  • 启动消息消费服务。这里只是启动消费服务，但是没有启动开始消费消息。
  • 绑定消费者 group 和 消费者
  • 启动 MQ Client Instance. 这里，在生产者中也调用了 mQClientFactory.start(); 方法. 有个疑问，为什么消费者会启动消息推送服务呢? 因为在push模式下，消费超时的消息会重新发送给Broker。所以是会使用消息推送服务的.
  • 从NameServer拉取topic的订阅信息
  • 向Broker校验客户端
  • 向所有的Broker的master节点发送心跳包，并上传FilterClass源文件给FilterServer
  • 立即消费消息： 将当前consumer负载得到的MessageQueue全部添加到PullMessageService.pullRequestQueue(阻塞队列)然后PullMessageService服务会开始拉取消息。消费消息。
pull模式: DefaultLitePullConsumerImpl.start()

先看下 DefaultLitePullConsumer 的一个简单使用:

DefaultLitePullConsumer 的实现代码要比 DefaultMQPushConsumer 的代码 规整很多。在 start() 方法中定义了很多的子方法进行调用。

  • 检查属性配置是否合法。
  • 初始化 MqClientInstance.
    • 创建 MQClientInstance 实例
    • 注册 consumerGroup 和 当前消费者的关系
  • 初始化 消息消费的 再平衡 服务。
  • 配置消费者再平衡的消费者组，消息模式，消息的分配策略， MQClientInstance 实例。
  • 实例化消息拉取的包装类 并注册消息过滤的钩子
  • 加载消息的消费偏移量。 如果是广播消息从本地获取偏移数据，如果是集群消息的话，则从远程获取偏移数据
  • 启动 MqClientInstance 实例
    • 设置 NameServer 的地址
    • 启动 remoteClient. 底层使用的通讯框架是Netty，提供了实现类NettyRemotingClient
    • 开启定时任务
    • 启动拉取消息的服务
    • 启动负载均衡服务
    • 启动推送消息的服务
    • 修改服务的状态为启动成功
  • 启动后操作
    • 如果是广播的模式：更新topic的订阅关系
    • 更新消息拉取任务
    • 拉取Topic的messageQueue.
    • 检查 Broker Client
Pull 模式 和 push 模式的对比

现在去谈push 和 pull 两种模式的对比，还为时尚早，我就先从启动上来看下，两种方式启动的不同点:

  • pull 模式没有拷贝订阅关系，也就是说 pull 模式下，RocketMQ 是没有提供重投机制的。
  • pull 模式没有和 Broker 保持心跳包。 如果消费者过多的时候，push 模式必然会对Broker造成比较啊的压力。

消费者总结

RocketMQ 对 pull 和 push 两种消息的消费模式提供了支持。pull 模式，对应的实现是 DefaultMqLitePullConsumer。 push 模式对应的实现是 DefaultMqPushConsumer. 仅仅对启动过程，两者的启动过程稍有不同。Push 模式下，RocketMQ 封装了 消息消费的重投机制，pull模式则没有，一切都需要消费者自己去实现。 push 模式会把 根据 MessageQueue的分配策略，将MessageQueue拉取到本地，存储到阻塞队列中，然后通过回调消费者注册监听器进行消费。 pull 模式则通过在消费逻辑中定时的轮询获取消息进行消费。 下一篇文章，我会仔细的分析消息消费的过程。

不管是Pull模式，还是push模式，在启动过程都是 创建 MqClientInstance 实例，并启动。

最后

期望和你一起遇见更好的自己

配套资料 · 免费领取
本文属于「BACKEND」资料包范围：中间件源码阅读路线图。按 RocketMQ / Redis / MySQL 分章节的完整路线。
在公众号「方家小白」回复关键词 BACKEND 即可领取。全文无需关注即可阅读。
BACKEND
版权与引用：本文由 方家小白 原创，首发于 https://fangjiaxiaobai.github.io/2021/09/28/rocketMQ/2-01-RocketMQ-Arch-Start/。 采用 CC BY-NC-SA 4.0 许可：允许非商业性转载与引用，须保留作者署名与原文链接，禁止全文复制用于商业用途或模型训练语料。
内容指纹：3c4b9a7e893988fc（用于溯源与侵权举证）。