RocketMQ(4.8.0)——RocketMQ的路由原理

RocketMQ的路由原理

　　生产者发送消息、消费者消费时都需要从 Namesrv 拉取 Topic 路由信息，那么这些路由信息是如何注册到 Namesrv 的呢？如果 Broker 异常宕机，路由信息又是如何更新的呢？

一、路由注册

　　Namesrv 获取的 Topic 路由信息来自 Broker 定时心跳，心跳时 Broker 将 Topic 信息和其他信息发送到 Namesrv。Namesrv 通过 RequestCode.REGISTER_BROKER 接口将心跳中的 Broker 信息和 Topic 信息存储在 Namesrv 中。

　　Namesrv 接收请求后，以 3.0.11 版本作为分水岭，按照版本分别做不同的处理，代码路径：D: ocketmq-master amesrvsrcmainjavaorgapache ocketmq amesrvprocessorDefaultRequestProcessor.java，代码如下：

   @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;
    }

　　因为当前 RocketMQ 的版本为 4.8.0，所以会 return this.registerBrokerWithFilterServer(ctx, request)，我们主要看这个方法中的 this.namesrvController.getRouteInfoManager().registerBroker() 方法，该方法的主要功能是将 request 解码为路由对象，保存在 Namesrv 内存中。

　　在路由信息注册完成后，Broker 会每隔 30s 发送一个注册请求给集群中全部的 Namesrv，俗称心跳信，会把最新的 Topic 路由信息注册到 Namesrv 中。

二、路由剔除

　　如果 Broker 长久没有心跳或者宕机，那么 Namesrv 会将这些不提供服务的 Broker 剔除。同时生产者和消费者在与 Namesrv 心跳后也会感知到被踢掉的 Broker，如此 Broker扩容或者宕机对生产、消费无感知的情况就处理完了。

　　Namesrv 有两种剔除 Broker的方式：

第一种：Broker 主动关闭时，会调用 Namesrv 的取消注册 Broker 的接口 RequestCode = RequestCode.UNREGISTER_BROKER，将自身从集群中删除。这个过程与路由注册的过程相反。

第二种：Namesrv 通过定时扫描已经下线的 Broker，将其主动剔除，实现过程代码路径：D: ocketmq-master amesrvsrcmainjavaorgapache ocketmq amesrvNamesrvController.java 中initialize()方法，代码如下：

public boolean initialize() {

        this.kvConfigManager.load();

        this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);

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

        this.registerProcessor();

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                NamesrvController.this.routeInfoManager.scanNotActiveBroker();
            }
        }, 5, 10, TimeUnit.SECONDS);

        this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                NamesrvController.this.kvConfigManager.printAllPeriodically();
            }
        }, 1, 10, TimeUnit.MINUTES);

        if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
            // Register a listener to reload SslContext
            try {
                fileWatchService = new FileWatchService(
                    new String[] {
                        TlsSystemConfig.tlsServerCertPath,
                        TlsSystemConfig.tlsServerKeyPath,
                        TlsSystemConfig.tlsServerTrustCertPath
                    },
                    new FileWatchService.Listener() {
                        boolean certChanged, keyChanged = false;
                        @Override
                        public void onChanged(String path) {
                            if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) {
                                log.info("The trust certificate changed, reload the ssl context");
                                reloadServerSslContext();
                            }
                            if (path.equals(TlsSystemConfig.tlsServerCertPath)) {
                                certChanged = true;
                            }
                            if (path.equals(TlsSystemConfig.tlsServerKeyPath)) {
                                keyChanged = true;
                            }
                            if (certChanged && keyChanged) {
                                log.info("The certificate and private key changed, reload the ssl context");
                                certChanged = keyChanged = false;
                                reloadServerSslContext();
                            }
                        }
                        private void reloadServerSslContext() {
                            ((NettyRemotingServer) remotingServer).loadSslContext();
                        }
                    });
            } catch (Exception e) {
                log.warn("FileWatchService created error, can't load the certificate dynamically");
            }
        }

        return true;
    }

　　这里定时执行 scanNotActiveBroker()，代码路径：D: ocketmq-master amesrvsrcmainjavaorgapache ocketmq amesrv outeinfoRouteInfoManager.java，代码如下：

    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();
            if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
                RemotingUtil.closeChannel(next.getValue().getChannel());
                it.remove();
                log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
                this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
            }
        }
    }

　　该方法会扫描全部已经注册的 Broker，依次将每一个 Broker 心跳的最后更新时间和当前时间做对比，如果 Broker 心跳的最后更新时间超过 BROKER_CHANNEL_EXPIRED_TIME(1000 * 60 * 2 = 120s)，则将 Broker 剔除。至此没有心跳的 Broker 从路由中被剔除，而客户端无任何感知。