RocketMQ-02

发表于 | 分类于

RocketMQ笔记（二）—— NameServer

谨记 RocketMQ 的设计原则—— 简单但高效

RocketMQ 采用发布订阅模式，基本的参与组件主要包括四部分：路由发现 (namesrv)、消息发送者 (producer)、消息服务器 (broker)、消息消费者 (consumer)

本文主要介绍 RocketMQ 的 路由管理 、 服务注册 、 服务发现 机制，由内部 NameServer 实现

架构

基于 RocketMQ 的设计理念，其摒弃了业界常用的注册中心 Zookeeper，而采用自研的 NameServer 来实现元数据的管理 (Broker/Topic/Queue 等信息)

由于 RocketMQ Topic 等原数据无需在集群之间保持强一致性，最终一致性即可，且能容忍分钟级别的数据不一致，NameServer 集群之间并不进行通信，极大降低其实现复杂性，对网络要求也降低不少，从而性能相较于 Zookeeper 提升很多

RocketMQ 物理部署图

NameServer 中的路由信息由 Broker 启动时注册，一个 Broker 启动时与所有 NameServer 建立一个 长连接，并将信息注册到上面，之后 NameServer 每10秒检测 Broker (扫描元数据 brokerLiveTable 的最后收到心跳包时间)是否存活，如果检测到 Broker 宕机，则将其信息从路由表中剔除，但剔除后并不会马上通知生产者

注意 Brocker 集群，这里简单将其区分为内外集群，上述部署图中，Broker Master1、Broker Slave1 数据一个内部主从集群，其内部的队列、消息一致，Broker1、Broker2 数据外部同级集群，有点绕。。。。

消息的发送

消息的生产者会在发送消息前从 NameServer 中获取 Broker 的服务器地址列表，然后通过负载均衡挑选一台服务器进行消息发送

路由元数据

路由实现类 org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager，存储的元数据如下

1
2
3
4
5
private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

路由注册

Broker 在启动时会先与集群中所有 NameServer 建立一个长连接并发送注册请求，之后每隔 30秒 向所有 NameServer 发送心跳包，NameServer 在接收到心跳包时会去更新 brokerLiveTable 的 lastUpdateTimestamp ，即最近接收到此 Broker 发出心跳包的时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

    @Override
    public void run() {
        try {
            BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
        } catch (Throwable e) {
            log.error("registerBrokerAll Exception", e);
        }
    }
}, 1000 * 10, 
        Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), 
        TimeUnit.MILLISECONDS);
// 发送间隔默认为 30s，可在配置文件中配置
private int registerNameServerPeriod = 1000 * 30;

NameServer 每隔 10s 会去扫描 brokerLiveTable，检查到 120s 都没有收到心跳包的 Broker，NameServer 将会移除该 Broker 的路由信息同时关闭 Socket 连接

路由删除

两种方式出发路由删除

  • 120s 未收到某 Borker 的心跳包
  • Broker 在正常关闭的情况下，会执行 unregisterBroker 进行路由删除