RocketMQ源码分析之路由中心

微信公众号「后端进阶」，专注后端技术分享：Java、Golang、WEB框架、分布式中间件、服务治理等等。
老司机倾囊相授，带你一路进阶，来不及解释了快上车！

早期的rocketmq版本的路由功能是使用zookeeper实现的，后来rocketmq为了追求性能，自己实现了一个性能更高效且实现简单的路由中心NameServer，而且可以通过部署多个路由节点实现高可用，但它们之间并不能互相通信，这也就会导致在某一个时刻各个路由节点间的数据并不完全相同，但数据某个时刻不一致并不会导致消息发送不了，这也是rocketmq追求简单高效的一个做法。

路由启动

看了Nameserver的源码后大呼惊叹，整个NameServer总共就由这么几个类类组成：

rocketmq nameserver

其中NamesrvStartup为启动类，NamesrvController为核心控制器，RouteInfoManager为路由信息表。

知道了这几个类的功能之后，我们就直接定位到NamesrvStartup启动类的启动方法：

org.apache.rocketmq.namesrv.NamesrvStartup#main0：

public static NamesrvController main0(String[] args) {
    try {
        // 步骤一
        NamesrvController controller = createNamesrvController(args);
        // 步骤二
        start(controller);
        String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();
        log.info(tip);
        System.out.printf("%s%n", tip);
        return controller;
    } catch (Throwable e) {
        e.printStackTrace();
        System.exit(-1);
    }
    return null;
}

整个NameServer服务启动的流程代码都在main0(String[] args)方法了，看起来是不是很简单，我们继续往下撸它的具体实现：

步骤一：

org.apache.rocketmq.namesrv.NamesrvStartup#createNamesrvController：

这个方法的代码有点多，下面我会拆分成几段进行分析:

// 创建命令行参数对象，这里定义了 -h 和 -n参数
Options options = ServerUtil.buildCommandlineOptions(new Options());
// 根据Options和运行时参数args生成命令行对象，buildCommandlineOptions定义了-c参数（Name server config properties file）和-p参数（Print all config item）
commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());
if (null == commandLine) {
    System.exit(-1);
    return null;
}

这里使用了Apache Commons CLI 命令行解析工具，它可以帮助开发者快速构建启动命令，并且帮助你组织命令的参数、以及输出列表等。

这段主要是根据运行时传递的参数生成commandLine命令行对象，用于解析运行时类似于 -c 指定文件路径，然后填充到namesrvConfig和nettyServerConfig对象中。

final NamesrvConfig namesrvConfig = new NamesrvConfig();
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
nettyServerConfig.setListenPort(9876);
if (commandLine.hasOption('c')) {
    // 读取命令行-c参数指定的配置文件
    String file = commandLine.getOptionValue('c');
    if (file != null) {
        // 将文件转成输入流
        InputStream in = new BufferedInputStream(new FileInputStream(file));
        properties = new Properties();
        // 加载到属性对象
        properties.load(in);
        // 装载配置
        MixAll.properties2Object(properties, namesrvConfig);
        MixAll.properties2Object(properties, nettyServerConfig);
    }
}

这段是createNamesrvController(String[] args)方法最为核心的代码，从代码知道先创建NamesrvConfig和NettyServerConfig对象，接着利用commandLine命令行工具读取-c指定的配置文件路径，然后将其读取到流中，生成properties对象，最后将namesrvConfig和nettyServerConfig对象进行初始化。

final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

// remember all configs to prevent discard
controller.getConfiguration().registerConfig(properties);

到这里就水到渠成地利用namesrvConfig和nettyServerConfig对象创建NamesrvController对象，然后在注册一遍properties防止丢失。

createNamesrvController(String[] args)这一步也是启动nameserver最为关键的操作，它为我们启动时提供了namesrvConfig和nettyServerConfig配置对象，以及创建NamesrvController核心控制器。

步骤二：

org.apache.rocketmq.namesrv.NamesrvStartup#start：

public static NamesrvController start(final NamesrvController controller) throws Exception {

    if (null == controller) {
        throw new IllegalArgumentException("NamesrvController is null");
    }

    // 对核心控制器进行初始化操作
    boolean initResult = controller.initialize();
    if (!initResult) {
        controller.shutdown();
        System.exit(-3);
    }

    // 注册一个钩子函数，用于JVM进程关闭时，优雅地释放netty服务、线程池等资源
    Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
        @Override
        public Void call() throws Exception {
            controller.shutdown();
            return null;
        }
    }));

    // 核心控制器启动操作
    controller.start();

    return controller;
}

步骤二也是一个次级的启动流程控制方法，该方法主要对核心控制器进行初始化操作，同时注册一个钩子函数，用于JVM进程关闭时，优雅地释放netty服务、线程池等资源，最后对核心控制器进行启动操作，接下来我们继续撸详细实现：

org.apache.rocketmq.namesrv.NamesrvController#initialize：

public boolean initialize() {
    // 加载KV配置
    this.kvConfigManager.load();
    // 创建Netty网络服务对象
    this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);

    this.remotingExecutor =
        Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
    this.registerProcessor();

    // 创建定时任务--每个10s扫描一次Broker，并定时剔除不活跃的Broker
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            NamesrvController.this.routeInfoManager.scanNotActiveBroker();
        }
    }, 5, 10, TimeUnit.SECONDS);

    // 创建定时任务--每个10分钟打印一遍KV配置
    this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            NamesrvController.this.kvConfigManager.printAllPeriodically();
        }
    }, 1, 10, TimeUnit.MINUTES);

    // ...

    return true;
}

该方法主要是对核心控制器进行启动前的一些初始化操作，包括根据NamesrvConfig的kvConfigPath存储KV配置属性的路径加载KV配置，创建定时任务：每个10s扫描一次Broker，并定时剔除不活跃的Broker；每个10分钟打印一遍KV配置。

这里的每个10s扫描一次Broker，并定时剔除不活跃的Broker，这里是路由删除的一些逻辑，后面会讲到。

org.apache.rocketmq.namesrv.NamesrvController#start：

public void start() throws Exception {
    this.remotingServer.start();

    if (this.fileWatchService != null) {
        this.fileWatchService.start();
    }
}

到这里对启动进行最后一步操作，即创建Netty服务，我们知道rocketmq是通过netty来进行通信，对应netty的一些细节这里就不展开讲了，后面我也会计划写一些关于netty的系列文章，敬请期待。

路由启动时序图：

nameServer startup

路由注册

路由注册即是Broker向Nameserver注册的过程，它们是通过Broker的心跳功能实现的，那么既然Nameserver是用来存储Broker的注册信息，那么我们就先来看看Nameserver到底存储了哪些信息，回到文章最开始的那张结构图，我们知道RouteInfoManager为路由信息表：

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager：

public class RouteInfoManager {
    private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    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;
}
  • topicQueueTable：Topic消息队列路由信息，包括topic所在的broker名称，读队列数量，写队列数量，同步标记等信息，rocketmq根据topicQueueTable的信息进行负载均衡消息发送。
  • brokerAddrTable：Broker节点信息，包括brokername，所在集群名称，还有主备节点信息。
  • clusterAddrTable：Broker集群信息，存储了集群中所有的Brokername。
  • brokerLiveTable：Broker状态信息，Nameserver每次收到Broker的心跳包就会更新该信息。

这里也先讲一下rocketmq是基于订阅发布机制，我之前也写过一篇文章《rocketmq的消费模式》，我们可知一个Topic拥有多个消息队列，如果不指定队列的数量，一个Broker会为每个Topic创建4个读队列和4个写队列，多个Broker组成集群，Broker会通过发送心跳包将自己的信息注册到路由中心，路由中心brokerLiveTable存储Broker的状态，它会根据Broker的心跳包更新Broker状态信息。

步骤一：Broker发送心跳包

org.apache.rocketmq.broker.BrokerController#start：

public void start() throws Exception {
    
    // 初次启动，这里会强制执行发送心跳包
    this.registerBrokerAll(true, false, true);
    
    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);
}

Broker在核心控制器启动时，会强制发送一次心跳包，接着创建一个定时任务，定时向路由中心发送心跳包。

org.apache.rocketmq.broker.BrokerController#registerBrokerAll：

public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) {
    // 创建一个topic包装类
    TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();

    // 这里比较有趣，如果该broker没有读写权限，那么会新建一个临时的topicConfigTable，再set进包装类
    if (!PermName.isWriteable(this.getBrokerConfig().getBrokerPermission())
        || !PermName.isReadable(this.getBrokerConfig().getBrokerPermission())) {
        ConcurrentHashMap<String, TopicConfig> topicConfigTable = new ConcurrentHashMap<String, TopicConfig>();
        for (TopicConfig topicConfig : topicConfigWrapper.getTopicConfigTable().values()) {
            TopicConfig tmp =
                new TopicConfig(topicConfig.getTopicName(), topicConfig.getReadQueueNums(), topicConfig.getWriteQueueNums(),
                                this.brokerConfig.getBrokerPermission());
            topicConfigTable.put(topicConfig.getTopicName(), tmp);
        }
        topicConfigWrapper.setTopicConfigTable(topicConfigTable);
    }

     // 判断是否该Broker是否需要发送心跳包
    if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
                                      this.getBrokerAddr(),
                                      this.brokerConfig.getBrokerName(),
                                      this.brokerConfig.getBrokerId(),
                                      this.brokerConfig.getRegisterBrokerTimeoutMills())) {
        // 执行发送心跳包
        doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
    }
}

该方法是Broker执行发送心跳包的核心控制方法，它主要做了topic的包装类封装操作，判断Broker此时是否需要执行发送心跳包，但我查了下org.apache.rocketmq.common.BrokerConfig#forceRegister字段的值永远等于true，也就是该判断永远为true，即每次都需要发送心跳包。

我们定位到needRegister远程调用到路由中心的方法：

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#isBrokerTopicConfigChanged：

public boolean isBrokerTopicConfigChanged(final String brokerAddr, final DataVersion dataVersion) {
    DataVersion prev = queryBrokerTopicConfig(brokerAddr);
    return null == prev || !prev.equals(dataVersion);
}

public DataVersion queryBrokerTopicConfig(final String brokerAddr) {
    BrokerLiveInfo prev = this.brokerLiveTable.get(brokerAddr);
    if (prev != null) {
        return prev.getDataVersion();
    }
    return null;
}

发现，Broker是否需要发送心跳包由该Broker在路由中心org.apache.rocketmq.namesrv.routeinfo.BrokerLiveInfo#dataVersion决定，如果dataVersion为空或者当前dataVersion不等于brokerLiveTable存储的brokerLiveTable，Broker就需要发送心跳包。

步骤二：Nameserver处理心跳包

Nameserver的netty服务监听收到心跳包之后，会调用到路由中心以下方法进行处理：

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#registerBroker：

public RegisterBrokerResult registerBroker(
    final String clusterName,
    final String brokerAddr,
    final String brokerName,
    final long brokerId,
    final String haServerAddr,
    final TopicConfigSerializeWrapper topicConfigWrapper,
    final List<String> filterServerList,
    final Channel channel) {
    RegisterBrokerResult result = new RegisterBrokerResult();
    try {
        try {
            this.lock.writeLock().lockInterruptibly();

            // 获取集群下所有的Broker，并将当前Broker加入clusterAddrTable，由于brokerNames是Set结构，并不会重复
            Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
            if (null == brokerNames) {
                brokerNames = new HashSet<String>();
                this.clusterAddrTable.put(clusterName, brokerNames);
            }
            brokerNames.add(brokerName);

            boolean registerFirst = false;

            // 获取Broker信息，如果是首次注册，那么新建一个BrokerData并加入brokerAddrTable
            BrokerData brokerData = this.brokerAddrTable.get(brokerName);
            if (null == brokerData) {
                registerFirst = true;
                brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
                this.brokerAddrTable.put(brokerName, brokerData);
            }
            // 这里判断Broker是否是已经注册过
            String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
            registerFirst = registerFirst || (null == oldAddr);

            // 如果是Broker是Master节点吗，并且Topic信息更新或者是首次注册，那么创建更新topic队列信息
            if (null != topicConfigWrapper
                && MixAll.MASTER_ID == brokerId) {
                if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
                    || registerFirst) {
                    ConcurrentMap<String, TopicConfig> tcTable =
                        topicConfigWrapper.getTopicConfigTable();
                    if (tcTable != null) {
                        for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
                            this.createAndUpdateQueueData(brokerName, entry.getValue());
                        }
                    }
                }
            }

            // 更新BrokerLiveInfo状态信息
            BrokerLiveInfo prevBrokerLiveInfo = 
                this.brokerLiveTable.put(brokerAddr,new BrokerLiveInfo(System.currentTimeMillis(),topicConfigWrapper.getDataVersion(),channel,haServerAddr));
        } finally {
            this.lock.writeLock().unlock();
        }
    } catch (Exception e) {
        log.error("registerBroker Exception", e);
    }

    return result;
}

该方法是处理Broker心跳包的最核心方法，它主要做了对RouteInfoManager路由信息的一些更新操作，包括对clusterAddrTable、brokerAddrTable、topicQueueTable、brokerLiveTable等路由信息。

路由注册时序图：

Broker register

路由删除

前面部分我们分析了Nameserver启动时会创建一个定时任务，定时剔除不活跃的Broker。

org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager#scanNotActiveBroker：

public void scanNotActiveBroker() {
    Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
    while (it.hasNext()) {
        Entry<String, BrokerLiveInfo> next = it.next();
        long last = next.getValue().getLastUpdateTimestamp();
        // 如果当前时间大于最后修改时间加上Broker过期时间，那么就剔除该Broker
        if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
            // 关闭Broker对应的channel
            RemotingUtil.closeChannel(next.getValue().getChannel());
            it.remove();
            log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
            // 从brokerLiveTable、brokerAddrTable、topicQueueTable移除Broker相关信息
            this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
        }
    }
}

剔除Broker信息的逻辑比较简单，首先从BrokerLiveInfo获取状态信息，判断Broker的心跳时间是否已超过限定值，若超过之后就执行剔除逻辑。

分析完了rocketmq自带的路由中心源码，其实我们自己实现一个路由中心貌似也不难。有时候我们发现公司有些项目可以独立拆分出来做成中间件的形式，也就是单独部署，其它业务依赖client包调用中间件服务，比如短信、推送、邮件、配置等模块。如果我们把这些中间件做成高可用集群部署，也可以考虑自己实现一个路由中心。

公众号「后端进阶」，专注后端技术分享！
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 196,583评论 5赞 462
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 82,669评论 2赞 374
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 143,684评论 0赞 325
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 52,682评论 1赞 267
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 61,533评论 5赞 358
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 46,396评论 1赞 275
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 36,814评论 3赞 387
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 35,458评论 0赞 255
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 39,745评论 1赞 294
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 34,789评论 2赞 314
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 36,565评论 1赞 326
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 32,410评论 3赞 315
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 37,828评论 3赞 300
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,050评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 30,342评论 1赞 253
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 41,793评论 2赞 342
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 41,010评论 2赞 337

推荐阅读更多精彩内容

  • 还记得你曾经最想活出的样子吗？
    ​ 只愿毕生棱角不平，傲气不减，从前往日，一如今朝。 [还记得你曾经最想活出的样子吗？] 讨论到这个话题的时候，会...
    五象限阅读 468评论 0赞 3
  • 我的2018-美好的一年
    2018年是我参加工作以后，过得最快的一年，时间像瀑布一样，一泻千里，把我们冲到了现在。总觉得小时候时间过得很慢，...
    小果果很好吃阅读 603评论 1赞 5
  • 冬思
    闲闲依孤窗， 清茶烈酒何物洒？ 飘飘是冰花。
    果z辉煌阅读 215评论 0赞 2
  • 用一生去怀念你
    这是我们一起走过的第四个年头了，时间它太快，而我走的太慢，我要以怎样的速度才能追赶上你世界里没有我的十九年？这...
    猪会飞123阅读 380评论 0赞 1
  • 每日一句
    The most rewarding things you do in life are often the on...
    可乐W阅读 233评论 0赞 0