一文读懂RocketMQ的高可用机制——集群管理高可用

public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {    System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));
    Options options = ServerUtil.buildCommandlineOptions(new Options());    commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());    if (null == commandLine) {        System.exit(-1);        return null;    }    // 创建NamesrvConfig    final NamesrvConfig namesrvConfig = new NamesrvConfig();    // 创建NettyServerConfig    final NettyServerConfig nettyServerConfig = new NettyServerConfig();    // 设置启动端口号    nettyServerConfig.setListenPort(9876);    // 解析启动-c参数    if (commandLine.hasOption('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);
            namesrvConfig.setConfigStorePath(file);
            System.out.printf("load config properties file OK, %s%n", file);            in.close();        }    }    // 解析启动-p参数    if (commandLine.hasOption('p')) {        InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);        MixAll.printObjectProperties(console, namesrvConfig);        MixAll.printObjectProperties(console, nettyServerConfig);        System.exit(0);    }    // 将启动参数填充到namesrvConfig,nettyServerConfig    MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
    if (null == namesrvConfig.getRocketmqHome()) {        System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);        System.exit(-2);    }
    LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();    JoranConfigurator configurator = new JoranConfigurator();    configurator.setContext(lc);    lc.reset();    configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");
    log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    MixAll.printObjectProperties(log, namesrvConfig);    MixAll.printObjectProperties(log, nettyServerConfig);    // 创建NameServerController    final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);
    // remember all configs to prevent discard    controller.getConfiguration().registerConfig(properties);
    return controller;}

public boolean initialize() {    // 加载KV配置    this.kvConfigManager.load();    // 创建NettyServer网络处理对象    this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);    // 开启定时任务:每隔10s扫描一次Broker,移除不活跃的Broker    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);    // 开启定时任务:每隔10min打印一次KV配置    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;}

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钩子函数代码    Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {        @Override        public Void call() throws Exception {            // 释放资源            controller.shutdown();            return null;        }    }));
    controller.start();
    return controller;}

// org.apache.rocketmq.remoting.netty.NettyRequestProcessorpublic interface NettyRequestProcessor {    RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)        throws Exception;
    boolean rejectRequest();}

public List<RegisterBrokerResult> registerBrokerAll(    final String clusterName,    final String brokerAddr,    final String brokerName,    final long brokerId,    final String haServerAddr,    final TopicConfigSerializeWrapper topicConfigWrapper,    final List<String> filterServerList,    final boolean oneway,    final int timeoutMills,    final boolean compressed) {
    final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();    // 获得nameServer地址信息    List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();    // 遍历所有nameserver列表    if (nameServerAddressList != null && nameServerAddressList.size() > 0) {        // 封装请求头        final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();        requestHeader.setBrokerAddr(brokerAddr);        requestHeader.setBrokerId(brokerId);        requestHeader.setBrokerName(brokerName);        requestHeader.setClusterName(clusterName);        requestHeader.setHaServerAddr(haServerAddr);        requestHeader.setCompressed(compressed);        // 封装请求体        RegisterBrokerBody requestBody = new RegisterBrokerBody();        requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);        requestBody.setFilterServerList(filterServerList);        final byte[] body = requestBody.encode(compressed);        final int bodyCrc32 = UtilAll.crc32(body);        requestHeader.setBodyCrc32(bodyCrc32);        final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());        for (final String namesrvAddr : nameServerAddressList) {            brokerOuterExecutor.execute(new Runnable() {                @Override                public void run() {                    try {                        // 分别向NameServer注册                        RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);                        if (result != null) {                            registerBrokerResultList.add(result);                        }
                        log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);                    } catch (Exception e) {                        log.warn("registerBroker Exception, {}", namesrvAddr, e);                    } finally {                        countDownLatch.countDown();                    }                }            });        }
        try {            countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);        } catch (InterruptedException e) {        }    }
    return registerBrokerResultList;}

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();            // 维护clusterAddrTable            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;            // 维护brokerAddrTable            BrokerData brokerData = this.brokerAddrTable.get(brokerName);            // 第一次注册,则创建brokerData            if (null == brokerData) {                registerFirst = true;                brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());                this.brokerAddrTable.put(brokerName, brokerData);            }            // 非第一次注册,更新Broker            Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();            //Switch slave to master: first remove <1, IP:PORT> in namesrv, then add <0, IP:PORT>            //The same IP:PORT must only have one record in brokerAddrTable            Iterator<Entry<Long, String>> it = brokerAddrsMap.entrySet().iterator();            while (it.hasNext()) {                Entry<Long, String> item = it.next();                if (null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey()) {                    it.remove();                }            }
            String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);            registerFirst = registerFirst || (null == oldAddr);            // 维护topicQueueTable            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());                        }                    }                }            }            // 维护brokerLiveTable            BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddr,                new BrokerLiveInfo(                    System.currentTimeMillis(),                    topicConfigWrapper.getDataVersion(),                    channel,                    haServerAddr));            if (null == prevBrokerLiveInfo) {                log.info("new broker registered, {} HAServer: {}", brokerAddr, haServerAddr);            }            // 维护filterServerList            if (filterServerList != null) {                if (filterServerList.isEmpty()) {                    this.filterServerTable.remove(brokerAddr);                } else {                    this.filterServerTable.put(brokerAddr, filterServerList);                }            }
            if (MixAll.MASTER_ID != brokerId) {                String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);                if (masterAddr != null) {                    BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.get(masterAddr);                    if (brokerLiveInfo != null) {                        result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());                        result.setMasterAddr(masterAddr);                    }                }            }        } finally {            this.lock.writeLock().unlock();        }    } catch (Exception e) {        log.error("registerBroker Exception", e);    }
    return result;}

public void scanNotActiveBroker() {  // 获得brokerLiveTable  Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();  // 遍历brokerLiveTable  while (it.hasNext()) {      Entry<String, BrokerLiveInfo> next = it.next();      long last = next.getValue().getLastUpdateTimestamp();      // 如果收到心跳包的时间距当时时间是否超过120s      if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {          // 关闭连接          RemotingUtil.closeChannel(next.getValue().getChannel());          // 移除broker          it.remove();          log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);          // 维护路由表          this.onChannelDestroy(next.getKey(), next.getValue().getChannel());      }  }}

public void onChannelDestroy(String remoteAddr, Channel channel) {    String brokerAddrFound = null;    if (channel != null) {        try {            try {                this.lock.readLock().lockInterruptibly();                Iterator<Entry<String, BrokerLiveInfo>> itBrokerLiveTable =                    this.brokerLiveTable.entrySet().iterator();                while (itBrokerLiveTable.hasNext()) {                    Entry<String, BrokerLiveInfo> entry = itBrokerLiveTable.next();                    if (entry.getValue().getChannel() == channel) {                        brokerAddrFound = entry.getKey();                        break;                    }                }            } finally {                this.lock.readLock().unlock();            }        } catch (Exception e) {            log.error("onChannelDestroy Exception", e);        }    }
    if (null == brokerAddrFound) {        brokerAddrFound = remoteAddr;    } else {        log.info("the broker's channel destroyed, {}, clean it's data structure at once", brokerAddrFound);    }
    if (brokerAddrFound != null && brokerAddrFound.length() > 0) {
        try {            try {                // 申请写锁,根据brokerAddress从brokerLiveTable和filterServerTable移除                this.lock.writeLock().lockInterruptibly();                this.brokerLiveTable.remove(brokerAddrFound);                this.filterServerTable.remove(brokerAddrFound);                // 维护brokerAddrTable                String brokerNameFound = null;                boolean removeBrokerName = false;                Iterator<Entry<String, BrokerData>> itBrokerAddrTable =                    this.brokerAddrTable.entrySet().iterator();                // 遍历brokerAddrTable                while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {                    BrokerData brokerData = itBrokerAddrTable.next().getValue();                    // 遍历broker地址                    Iterator<Entry<Long, String>> it = brokerData.getBrokerAddrs().entrySet().iterator();                    while (it.hasNext()) {                        Entry<Long, String> entry = it.next();                        Long brokerId = entry.getKey();                        String brokerAddr = entry.getValue();                        // 根据broker地址移除brokerAddr                        if (brokerAddr.equals(brokerAddrFound)) {                            brokerNameFound = brokerData.getBrokerName();                            it.remove();                            log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",                                brokerId, brokerAddr);                            break;                        }                    }                    // 如果当前主题只包含待移除的broker,则移除该topic                    if (brokerData.getBrokerAddrs().isEmpty()) {                        removeBrokerName = true;                        itBrokerAddrTable.remove();                        log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",                            brokerData.getBrokerName());                    }                }                // 维护clusterAddrTable                if (brokerNameFound != null && removeBrokerName) {                    Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();                    // 遍历clusterAddrTable                    while (it.hasNext()) {                        Entry<String, Set<String>> entry = it.next();                        // 获得集群名称                        String clusterName = entry.getKey();                        // 获得集群中brokerName集合                        Set<String> brokerNames = entry.getValue();                        // 从brokerNames中移除brokerNameFound                        boolean removed = brokerNames.remove(brokerNameFound);                        if (removed) {                            log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",                                brokerNameFound, clusterName);
                            if (brokerNames.isEmpty()) {                                log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",                                    clusterName);                                // 如果集群中不包含任何broker,则移除该集群                                it.remove();                            }
                            break;                        }                    }                }                // 维护topicQueueTable队列                if (removeBrokerName) {                    // 遍历topicQueueTable                    Iterator<Entry<String, List<QueueData>>> itTopicQueueTable =                        this.topicQueueTable.entrySet().iterator();                    while (itTopicQueueTable.hasNext()) {                        Entry<String, List<QueueData>> entry = itTopicQueueTable.next();                        // 主题名称                        String topic = entry.getKey();                        // 队列集合                        List<QueueData> queueDataList = entry.getValue();                        // 遍历该主题队列                        Iterator<QueueData> itQueueData = queueDataList.iterator();                        while (itQueueData.hasNext()) {                            // 从队列中移除为活跃broker信息                            QueueData queueData = itQueueData.next();                            if (queueData.getBrokerName().equals(brokerNameFound)) {                                itQueueData.remove();                                log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",                                    topic, queueData);                            }                        }                        // 如果该topic的队列为空,则移除该topic                        if (queueDataList.isEmpty()) {                            itTopicQueueTable.remove();                            log.info("remove topic[{}] all queue, from topicQueueTable, because channel destroyed",                                topic);                        }                    }                }            } finally {                // 释放写锁                this.lock.writeLock().unlock();            }        } catch (Exception e) {            log.error("onChannelDestroy Exception", e);        }    }}

public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,    RemotingCommand request) throws RemotingCommandException {    final RemotingCommand response = RemotingCommand.createResponseCommand(null);    final GetRouteInfoRequestHeader requestHeader =        (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);    // 调用RouteInfoManager的方法,从路由表topicQueueTable、brokerAddrTable、 filterServerTable中    // 分别填充TopicRouteData的List<QueueData>、List<BrokerData>、 filterServer    TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());    // 如果找到主题对应你的路由信息并且该主题为顺序消息，则从NameServer KVConfig中获取 关于顺序消息相关的配置填充路由信息    if (topicRouteData != null) {        if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {            String orderTopicConf =                this.namesrvController.getKvConfigManager().getKVConfig(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG,                    requestHeader.getTopic());            topicRouteData.setOrderTopicConf(orderTopicConf);        }
        byte[] content = topicRouteData.encode();        response.setBody(content);        response.setCode(ResponseCode.SUCCESS);        response.setRemark(null);        return response;    }
    response.setCode(ResponseCode.TOPIC_NOT_EXIST);    response.setRemark("No topic route info in name server for the topic: " + requestHeader.getTopic()        + FAQUrl.suggestTodo(FAQUrl.APPLY_TOPIC_URL));    return response;}