RocketMq4.7源码解析之三(NameServer 路由注册,删除,发现)

我就是我 2022-12-01 05:24 182阅读 0赞

### 文章目录 ###

*  NameServer 路由注册、故障剔除
 *  路由注册
 *   *  broker
     *  namesrv
     *   *  RouteInfoManager成员变量
         *  创建或更新TopicConfig
 *  路由删除
 *   *  定时检测失效删除
     *  主动关闭删除
     *   *  broker
         *  name
 *  路由发现

# NameServer 路由注册、故障剔除 #

名字服务（Name Server） 名称服务充当路由消息的提供者。生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。多个Namesrv实例组成集群，但相互独立，没有信息交换。

代理服务器（Broker Server）  
消息中转角色，负责存储消息、转发消息。代理服务器在RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备。代理服务器也存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center]

# 路由注册 #

Broker 启动时向集群中所有的NameServ发送心跳语句，每隔3 0s 向集群中所有NameServer 发送心跳包， NameServer 收到Broker 心跳包时会更新brokerLiveTable 缓存中BrokerLivelnfo 的lastUpdateTimestamp ，然后NameServer 每隔10s 扫描brokerLiveTable ，如果连续120s 没有收到心跳包， NameServer 将移除该Broker 的路由信息同时关闭Socket 连接。

## broker ##

broker为什么会向所有name注册了？是因为name之间不互相通讯.  
org.apache.rocketmq.broker.BrokerController\#start  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 1]

public synchronized void registerBrokerAll(final boolean checkOrderConfig, boolean oneway, boolean forceRegister) {
        
            //封装所有topic以及版本信息
            TopicConfigSerializeWrapper topicConfigWrapper = this.getTopicConfigManager().buildTopicConfigSerializeWrapper();
            //broker权限若不支持写读
            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);
            }
    
            if (forceRegister || needRegister(this.brokerConfig.getBrokerClusterName(),
                this.getBrokerAddr(),
                this.brokerConfig.getBrokerName(),
                this.brokerConfig.getBrokerId(),
                this.brokerConfig.getRegisterBrokerTimeoutMills())) {
        
                //向name注册所有topic
                doRegisterBrokerAll(checkOrderConfig, oneway, topicConfigWrapper);
            }
        }
    
    
    private void doRegisterBrokerAll(boolean checkOrderConfig, boolean oneway,
            TopicConfigSerializeWrapper topicConfigWrapper) {
        
            //向name注册所有topic
            List<RegisterBrokerResult> registerBrokerResultList = this.brokerOuterAPI.registerBrokerAll(
                this.brokerConfig.getBrokerClusterName(),
                this.getBrokerAddr(),
                this.brokerConfig.getBrokerName(),
                this.brokerConfig.getBrokerId(),
                this.getHAServerAddr(),
                topicConfigWrapper,
                this.filterServerManager.buildNewFilterServerList(),
                oneway,
                this.brokerConfig.getRegisterBrokerTimeoutMills(),
                this.brokerConfig.isCompressedRegister());
    
            if (registerBrokerResultList.size() > 0) {
        
                RegisterBrokerResult registerBrokerResult = registerBrokerResultList.get(0);
                if (registerBrokerResult != null) {
        
                    if (this.updateMasterHAServerAddrPeriodically && registerBrokerResult.getHaServerAddr() != null) {
        
                        this.messageStore.updateHaMasterAddress(registerBrokerResult.getHaServerAddr());
                    }
    
                    this.slaveSynchronize.setMasterAddr(registerBrokerResult.getMasterAddr());
    
                    if (checkOrderConfig) {
        
                        this.getTopicConfigManager().updateOrderTopicConfig(registerBrokerResult.getKvTable());
                    }
                }
            }
        }

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();
            //获取所有name地址
            List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
            if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
        
    
                //封装请求头
                final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
                requestHeader.setBrokerAddr(brokerAddr);
                requestHeader.setBrokerId(brokerId);
                requestHeader.setBrokerName(brokerName);
                requestHeader.setClusterName(clusterName);
                //master 地址，初次请求时该值为空， slave 向Nameserver 注册后返回。
                requestHeader.setHaServerAddr(haServerAddr);
                requestHeader.setCompressed(compressed);
    
                RegisterBrokerBody requestBody = new RegisterBrokerBody();
                /*
                主题配置,Broker中Topic默认存储在$｛ RocketMQ｝ ／store/confg/topic.json 中。
                 */
                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());
                //遍历所有name,多线程进行注册,闭锁用来同步执行结果
                for (final String namesrvAddr : nameServerAddressList) {
        
                    brokerOuterExecutor.execute(new Runnable() {
        
                        @Override
                        public void run() {
        
                            try {
        
                                //Broker启动时向集群中所有的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;
        }
    
    
      private RegisterBrokerResult registerBroker(
            final String namesrvAddr,
            final boolean oneway,
            final int timeoutMills,
            final RegisterBrokerRequestHeader requestHeader,
            final byte[] body
        ) throws RemotingCommandException, MQBrokerException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException,
            InterruptedException {
        
            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.REGISTER_BROKER, requestHeader);
            request.setBody(body);
    
            //是否单向发送
            if (oneway) {
        
                try {
        
                    //仅仅维持心跳,就直接返回
                    this.remotingClient.invokeOneway(namesrvAddr, request, timeoutMills);
                } catch (RemotingTooMuchRequestException e) {
        
                    // Ignore
                }
                return null;
            }
    
            //同步发送
            RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMills);
            assert response != null;
            switch (response.getCode()) {
        
                //访问成功处理
                case ResponseCode.SUCCESS: {
        
                    //解码responseHeader
                    RegisterBrokerResponseHeader responseHeader =
                        (RegisterBrokerResponseHeader) response.decodeCommandCustomHeader(RegisterBrokerResponseHeader.class);
                    //记录请求返回结果,并封装返回
                    RegisterBrokerResult result = new RegisterBrokerResult();
                    result.setMasterAddr(responseHeader.getMasterAddr());
                    result.setHaServerAddr(responseHeader.getHaServerAddr());
                    if (response.getBody() != null) {
        
                        result.setKvTable(KVTable.decode(response.getBody(), KVTable.class));
                    }
                    return result;
                }
                default:
                    break;
            }
    
            throw new MQBrokerException(response.getCode(), response.getRemark());
        }

@Override
        public RemotingCommand invokeSync(String addr, final RemotingCommand request, long timeoutMillis)
            throws InterruptedException, RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException {
        
            long beginStartTime = System.currentTimeMillis();
            //获取缓存通道
            final Channel channel = this.getAndCreateChannel(addr);
            //若通道处于活动状态
            if (channel != null && channel.isActive()) {
        
                try {
        
                    //调用前置钩子方法
                    doBeforeRpcHooks(addr, request);
                    long costTime = System.currentTimeMillis() - beginStartTime;
                    //超时处理
                    if (timeoutMillis < costTime) {
        
                        throw new RemotingTimeoutException("invokeSync call timeout");
                    }
                    //重新计算剩余超时时间
                    RemotingCommand response = this.invokeSyncImpl(channel, request, timeoutMillis - costTime);
                    //调用RPC后置方法
                    doAfterRpcHooks(RemotingHelper.parseChannelRemoteAddr(channel), request, response);
                    return response;
                } catch (RemotingSendRequestException e) {
        
                    log.warn("invokeSync: send request exception, so close the channel[{}]", addr);
                    this.closeChannel(addr, channel);
                    throw e;
                } catch (RemotingTimeoutException e) {
        
                    //客户端超时直接关闭socket通道
                    if (nettyClientConfig.isClientCloseSocketIfTimeout()) {
        
                        this.closeChannel(addr, channel);
                        log.warn("invokeSync: close socket because of timeout, {}ms, {}", timeoutMillis, addr);
                    }
                    log.warn("invokeSync: wait response timeout exception, the channel[{}]", addr);
                    throw e;
                }
            } else {
        
                //关闭通道
                this.closeChannel(addr, channel);
                throw new RemotingConnectException(addr);
            }
        }

public RemotingCommand invokeSyncImpl(final Channel channel, final RemotingCommand request,
            final long timeoutMillis)
            throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException {
        
            final int opaque = request.getOpaque();
    
            try {
        
                final ResponseFuture responseFuture = new ResponseFuture(channel, opaque, timeoutMillis, null, null);
                //缓存请求
                this.responseTable.put(opaque, responseFuture);
                //获取访问地址
                final SocketAddress addr = channel.remoteAddress();
                //注册一个 ChannelFutureListener，以便在操作完成时获得通知
                channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
        
                    @Override
                    public void operationComplete(ChannelFuture f) throws Exception {
        
                        //检查操作的状态
                        if (f.isSuccess()) {
        
                            responseFuture.setSendRequestOK(true);
                            return;
                        } else {
        
                            responseFuture.setSendRequestOK(false);
                        }
                        //移除访问请求
                        responseTable.remove(opaque);
                        responseFuture.setCause(f.cause());
                        responseFuture.putResponse(null);
                        log.warn("send a request command to channel <" + addr + "> failed.");
                    }
                });
    
                //使用闭锁等待超时时间,当通道接收消息后,会调用read0,
                // 然后调用本类processResponseCommand,通过opaque从 responseTable获取请求,
                //然后将responseCommand注入进去
                RemotingCommand responseCommand = responseFuture.waitResponse(timeoutMillis);
                //返回值为null
                if (null == responseCommand) {
        
                    //如果成功请求返回,则说明超时
                    if (responseFuture.isSendRequestOK()) {
        
                        throw new RemotingTimeoutException(RemotingHelper.parseSocketAddressAddr(addr), timeoutMillis,
                            responseFuture.getCause());
                    } else {
        
                        //否则发送异常
                        throw new RemotingSendRequestException(RemotingHelper.parseSocketAddressAddr(addr), responseFuture.getCause());
                    }
                }
                //返回响应结果
                return responseCommand;
            } finally {
        
                //移除访问请求
                this.responseTable.remove(opaque);
            }
        }

## namesrv ##

org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor\#processRequest  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 2]

public RemotingCommand registerBrokerWithFilterServer(ChannelHandlerContext ctx, RemotingCommand request)
            throws RemotingCommandException {
        
            //初始化默认返回结果
            final RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
            final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
            //request中解码RegisterBrokerRequestHeader
            final RegisterBrokerRequestHeader requestHeader =
                (RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);
    
            //校验数据完整性
            if (!checksum(ctx, request, requestHeader)) {
        
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("crc32 not match");
                return response;
            }
    
            RegisterBrokerBody registerBrokerBody = new RegisterBrokerBody();
    
            if (request.getBody() != null) {
        
                try {
        
                    //request中解码registerBrokerBody
                    registerBrokerBody = RegisterBrokerBody.decode(request.getBody(), requestHeader.isCompressed());
                } catch (Exception e) {
        
                    throw new RemotingCommandException("Failed to decode RegisterBrokerBody", e);
                }
            } else {
        
                //若请求体没有,则设置默认值
                registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setCounter(new AtomicLong(0));
                registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setTimestamp(0);
            }
    
            //注册broker路由信息
            RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
                requestHeader.getClusterName(),
                requestHeader.getBrokerAddr(),
                requestHeader.getBrokerName(),
                requestHeader.getBrokerId(),
                requestHeader.getHaServerAddr(),
                registerBrokerBody.getTopicConfigSerializeWrapper(),
                registerBrokerBody.getFilterServerList(),
                ctx.channel());
    
            //设置ha地址
            responseHeader.setHaServerAddr(result.getHaServerAddr());
            responseHeader.setMasterAddr(result.getMasterAddr());
    
            byte[] jsonValue = this.namesrvController.getKvConfigManager().getKVListByNamespace(NamesrvUtil.NAMESPACE_ORDER_TOPIC_CONFIG);
            response.setBody(jsonValue);
    
            //设置返回值
            response.setCode(ResponseCode.SUCCESS);
            //设置备注
            response.setRemark(null);
            return response;
        }

### RouteInfoManager成员变量 ###

private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
        //broker超时时间,超过则关闭channel
        private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
        private final ReadWriteLock lock = new ReentrantReadWriteLock();
        //Topic 消息队列路由信息，消息发送时根据路由表进行负载均衡。
        //一个Topic 拥有多个消息队列，一个Broker 为每一主题默认创建4 个读队列4 个写队列。
        private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
        //Broker 基础信息， 包含brokerName 、所属集群名称、主备Broker地址。
        private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
        //Broker 集群信息，存储集群中所有Broker 名称。
        private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
        //Broker 状态信息。NameServer 每次收到心跳包时会替换该信息。
        private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
        //Broker 上的FilterServer 列表，用于类模式消息过滤
        private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

/**
         * 设计亮点： Name Serve 与Broker 保持长连接， Broker 状态存储在brokerLiveTable 中，
         * NameS erver 每收到一个心跳包，将更新brokerL iveTa ble 中关于Broker 的状态信息以及路
         * 由表（ topicQueueTable 、brokerAddrTab le 、brokerLiveTable 、fi lterServerTable ） 。更新上述
         * 路由表（ HashTable ）使用了锁粒度较少的读写锁，允许多个消息发送者（P roducer ）并发读，
         * 保证消息发送时的高并发。但同一时刻NameServer 只处理一个Broker 心跳包，多个心跳
         * 包请求串行执行。这也是读写锁经典使用场景
         * @param clusterName
         * @param brokerAddr
         * @param brokerName
         * @param brokerId
         * @param haServerAddr
         * @param topicConfigWrapper
         * @param filterServerList
         * @param channel
         * @return
         */
     //路由注册
        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();
    
                    //获取集群名
                    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;
                    //集群获取brokerData数据
                    BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                    //若不存在,则新建brokerData
                    if (null == brokerData) {
        
                        //标记第一次创建
                        registerFirst = true;
                        brokerData = new BrokerData(clusterName, brokerName, new HashMap<Long, String>());
                        //放入brokerAddrTable
                        this.brokerAddrTable.put(brokerName, brokerData);
                    }
                    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
                    //主从切换
                    //同一条addr只能存在一条记录,删除存在的addr
                    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();
                        }
                    }
    
                    //添加新的addr和id
                    String oldAddr = brokerData.getBrokerAddrs().put(brokerId, brokerAddr);
                    registerFirst = registerFirst || (null == oldAddr);
    
                    if (null != topicConfigWrapper
                            //若broker为主
                        && MixAll.MASTER_ID == brokerId) {
        
                        //broker config配置信息变化
                        if (this.isBrokerTopicConfigChanged(brokerAddr, topicConfigWrapper.getDataVersion())
                                //或者第一次创建
                            || registerFirst) {
        
                            ConcurrentMap<String, TopicConfig> tcTable =
                                topicConfigWrapper.getTopicConfigTable();
                            if (tcTable != null) {
        
                                for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
        
                                    //创建或更新TopicConfig
                                    this.createAndUpdateQueueData(brokerName, entry.getValue());
                                }
                            }
                        }
                    }
    
                    //更新存活broker信息表,为删除路由重要依据
                    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);
                    }
    
                    //注册broker的过滤器Server地址列表
                    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) {
        
                                //获取ha和master的地址，返回
                                result.setHaServerAddr(brokerLiveInfo.getHaServerAddr());
                                result.setMasterAddr(masterAddr);
                            }
                        }
                    }
                } finally {
        
                    this.lock.writeLock().unlock();
                }
            } catch (Exception e) {
        
                log.error("registerBroker Exception", e);
            }
    
            return result;
        }

### 创建或更新TopicConfig ###

private void createAndUpdateQueueData(final String brokerName, final TopicConfig topicConfig) {
        
            QueueData queueData = new QueueData();
            queueData.setBrokerName(brokerName);
            queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
            queueData.setReadQueueNums(topicConfig.getReadQueueNums());
            queueData.setPerm(topicConfig.getPerm());
            queueData.setTopicSynFlag(topicConfig.getTopicSysFlag());
    
            List<QueueData> queueDataList = this.topicQueueTable.get(topicConfig.getTopicName());
            if (null == queueDataList) {
        
                queueDataList = new LinkedList<QueueData>();
                queueDataList.add(queueData);
                this.topicQueueTable.put(topicConfig.getTopicName(), queueDataList);
                log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);
            } else {
        
                boolean addNewOne = true;
                //若数据没变,则不处理,数据变更则删除后新增
                Iterator<QueueData> it = queueDataList.iterator();
                while (it.hasNext()) {
        
                    QueueData qd = it.next();
                    if (qd.getBrokerName().equals(brokerName)) {
        
                        if (qd.equals(queueData)) {
        
                            addNewOne = false;
                        } else {
        
                            log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), qd,
                                queueData);
                            it.remove();
                        }
                    }
                }
    
                if (addNewOne) {
        
                    queueDataList.add(queueData);
                }
            }
        }

# 路由删除 #

Name Server 会每隔I Os 扫描brokerLiveTable 状态表，如果BrokerLive 的lastUpdateTimestamp 的时间戳距当前时间超过120s ，则认为Broker 失效，移除该Broker,关闭与Broker 连接，并同时更新topicQueueTable 、brokerAddrTable 、brokerLive Table 、filterServerTable 。

RocktMQ 有两个触发点来触发路由删除。  
1 ) NameServer 定时扫描brokerLiveTable 检测上次心跳包与当前系统时间的时间差，如果时间戳大于120s ，则需要移除该Broker 信息。  
2 ) Broker 在正常被关闭的情况下，会执行unregisterBroker 指令。

## 定时检测失效删除 ##

org.apache.rocketmq.namesrv.NamesrvController\#initialize  
![在这里插入图片描述][20200826104118670.png_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 3]

public static void closeChannel(Channel channel) {
        
            //解析ip地址
            final String addrRemote = RemotingHelper.parseChannelRemoteAddr(channel);
            //注册一个 ChannelFutureListener，以便在操作完成时获得通知
            channel.close().addListener(new ChannelFutureListener() {
        
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
        
                    log.info("closeChannel: close the connection to remote address[{}] result: {}", addrRemote,
                        //检查操作的状态
                        future.isSuccess());
                }
            });
        }

/*
        移除步骤
            心跳信息->过滤信息->broker基础信息->broker集群信息->topic队列表
         */
        public void onChannelDestroy(String remoteAddr, Channel channel) {
        
            String brokerAddrFound = null;
            //通过通道获取broker地址
            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 {
        
                        this.lock.writeLock().lockInterruptibly();
                        //删除心跳信息
                        this.brokerLiveTable.remove(brokerAddrFound);
                        //删除过滤信息
                        this.filterServerTable.remove(brokerAddrFound);
                        String brokerNameFound = null;
                        boolean removeBrokerName = false;
                        Iterator<Entry<String, BrokerData>> itBrokerAddrTable =
                            this.brokerAddrTable.entrySet().iterator();
                        //遍历查找brokerAddrFound对应的brokerData,并从brokerData移除
                        while (itBrokerAddrTable.hasNext() && (null == brokerNameFound)) {
        
                            BrokerData brokerData = itBrokerAddrTable.next().getValue();
    
                            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();
                                if (brokerAddr.equals(brokerAddrFound)) {
        
                                    brokerNameFound = brokerData.getBrokerName();
                                    it.remove();
                                    log.info("remove brokerAddr[{}, {}] from brokerAddrTable, because channel destroyed",
                                        brokerId, brokerAddr);
                                    break;
                                }
                            }
    
                            //若brokerData不包含其他broker,则从brokerAddrTable移除
                            if (brokerData.getBrokerAddrs().isEmpty()) {
        
                                removeBrokerName = true;
                                itBrokerAddrTable.remove();
                                log.info("remove brokerName[{}] from brokerAddrTable, because channel destroyed",
                                    brokerData.getBrokerName());
                            }
                        }
    
                        if (brokerNameFound != null && removeBrokerName) {
        
                            Iterator<Entry<String, Set<String>>> it = this.clusterAddrTable.entrySet().iterator();
                            while (it.hasNext()) {
        
                                Entry<String, Set<String>> entry = it.next();
                                String clusterName = entry.getKey();
                                Set<String> brokerNames = entry.getValue();
                                //从集群中移除broker
                                boolean removed = brokerNames.remove(brokerNameFound);
                                if (removed) {
        
                                    log.info("remove brokerName[{}], clusterName[{}] from clusterAddrTable, because channel destroyed",
                                        brokerNameFound, clusterName);
    
                                    //若移除broker后,该集群为空,则从clusterAddrTable移除集群.
                                    if (brokerNames.isEmpty()) {
        
                                        log.info("remove the clusterName[{}] from clusterAddrTable, because channel destroyed and no broker in this cluster",
                                            clusterName);
                                        it.remove();
                                    }
    
                                    break;
                                }
                            }
                        }
    
                        if (removeBrokerName) {
        
                            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();
                                //移除brokerName对应的topic
                                while (itQueueData.hasNext()) {
        
                                    QueueData queueData = itQueueData.next();
                                    if (queueData.getBrokerName().equals(brokerNameFound)) {
        
                                        itQueueData.remove();
                                        log.info("remove topic[{} {}], from topicQueueTable, because channel destroyed",
                                            topic, queueData);
                                    }
                                }
    
                                //若brokerName对应的topic移除完,则从topicQueueTable移除该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);
                }
            }
        }

## 主动关闭删除 ##

### broker ###

org.apache.rocketmq.broker.BrokerController\#shutdown  
![在这里插入图片描述][2020082611573015.png_pic_center]

private void unregisterBrokerAll() {
        
            this.brokerOuterAPI.unregisterBrokerAll(
                this.brokerConfig.getBrokerClusterName(),
                this.getBrokerAddr(),
                this.brokerConfig.getBrokerName(),
                this.brokerConfig.getBrokerId());
        }

public void unregisterBrokerAll(
                final String clusterName,
                final String brokerAddr,
                final String brokerName,
                final long brokerId
            ) {
        
                //获取所有name地址
                List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
                if (nameServerAddressList != null) {
        
                    //依次通知每个name
                    for (String namesrvAddr : nameServerAddressList) {
        
                        try {
        
                            //关闭路由信息
                            this.unregisterBroker(namesrvAddr, clusterName, brokerAddr, brokerName, brokerId);
                            log.info("unregisterBroker OK, NamesrvAddr: {}", namesrvAddr);
                        } catch (Exception e) {
        
                            log.warn("unregisterBroker Exception, {}", namesrvAddr, e);
                        }
                    }
                }
            }

public void unregisterBroker(
            final String namesrvAddr,
            final String clusterName,
            final String brokerAddr,
            final String brokerName,
            final long brokerId
        ) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQBrokerException {
        
            //设置请求头
            UnRegisterBrokerRequestHeader requestHeader = new UnRegisterBrokerRequestHeader();
            requestHeader.setBrokerAddr(brokerAddr);
            requestHeader.setBrokerId(brokerId);
            requestHeader.setBrokerName(brokerName);
            requestHeader.setClusterName(clusterName);
            //发送UNREGISTER_BROKER请求
            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNREGISTER_BROKER, requestHeader);
    
            //执行
            RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, 3000);
            assert response != null;
            //成功处理
            switch (response.getCode()) {
        
                case ResponseCode.SUCCESS: {
        
                    return;
                }
                default:
                    break;
            }
    
            throw new MQBrokerException(response.getCode(), response.getRemark());
        }

### name ###

org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor\#processRequest

![在这里插入图片描述][20200826110914507.png_pic_center]

public RemotingCommand unregisterBroker(ChannelHandlerContext ctx,
            RemotingCommand request) throws RemotingCommandException {
        
            //响应头为null
            final RemotingCommand response = RemotingCommand.createResponseCommand(null);
            //解码请求头
            final UnRegisterBrokerRequestHeader requestHeader =
                (UnRegisterBrokerRequestHeader) request.decodeCommandCustomHeader(UnRegisterBrokerRequestHeader.class);
    
            //取消注册
            this.namesrvController.getRouteInfoManager().unregisterBroker(
                requestHeader.getClusterName(),
                requestHeader.getBrokerAddr(),
                requestHeader.getBrokerName(),
                requestHeader.getBrokerId());
    
            //设置响应码
            response.setCode(ResponseCode.SUCCESS);
            response.setRemark(null);
            return response;
        }

//路由删除1
        //Broker正常关闭
        public void unregisterBroker(
            final String clusterName,
            final String brokerAddr,
            final String brokerName,
            final long brokerId) {
        
            try {
        
                try {
        
                    this.lock.writeLock().lockInterruptibly();
                    //删除心跳信息
                    BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.remove(brokerAddr);
                    log.info("unregisterBroker, remove from brokerLiveTable {}, {}",
                        brokerLiveInfo != null ? "OK" : "Failed",
                        brokerAddr
                    );
    
                    //删除过滤信息
                    this.filterServerTable.remove(brokerAddr);
    
                    //删除broker信息
                    boolean removeBrokerName = false;
                    BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                    if (null != brokerData) {
        
                        String addr = brokerData.getBrokerAddrs().remove(brokerId);
                        log.info("unregisterBroker, remove addr from brokerAddrTable {}, {}",
                            addr != null ? "OK" : "Failed",
                            brokerAddr
                        );
    
                        if (brokerData.getBrokerAddrs().isEmpty()) {
        
                            this.brokerAddrTable.remove(brokerName);
                            log.info("unregisterBroker, remove name from brokerAddrTable OK, {}",
                                brokerName
                            );
    
                            removeBrokerName = true;
                        }
                    }
    
                    //删除集群信息
                    if (removeBrokerName) {
        
                        Set<String> nameSet = this.clusterAddrTable.get(clusterName);
                        if (nameSet != null) {
        
                            boolean removed = nameSet.remove(brokerName);
                            log.info("unregisterBroker, remove name from clusterAddrTable {}, {}",
                                removed ? "OK" : "Failed",
                                brokerName);
    
                            if (nameSet.isEmpty()) {
        
                                this.clusterAddrTable.remove(clusterName);
                                log.info("unregisterBroker, remove cluster from clusterAddrTable {}",
                                    clusterName
                                );
                            }
                        }
                        //删除主题信息
                        this.removeTopicByBrokerName(brokerName);
                    }
                } finally {
        
                    this.lock.writeLock().unlock();
                }
            } catch (Exception e) {
        
                log.error("unregisterBroker Exception", e);
            }
        }

# 路由发现 #

org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor\#processRequest  
![在这里插入图片描述][20200826121458335.png_pic_center]  
RocketMQ 路由发现是非实时的，当Topic 路由  
出现变化后， NameServer 不主动推送给客户端， 而  
是由客户端定时拉取主题最新的路由。根据主题名  
称拉取路由信息的命令编码为： GET ROUTEINTO  
BY\_TOPIC 。

//路由发现
        public RemotingCommand getRouteInfoByTopic(ChannelHandlerContext ctx,
            RemotingCommand request) throws RemotingCommandException {
        
            //响应头为null
            final RemotingCommand response = RemotingCommand.createResponseCommand(null);
            //解码请求头,获得topic信息
            final GetRouteInfoRequestHeader requestHeader =
                (GetRouteInfoRequestHeader) request.decodeCommandCustomHeader(GetRouteInfoRequestHeader.class);
    
            //填充TopicRouteData 中的List<QueueData＞、List<BrokerData＞和filterServer 地址表。
            TopicRouteData topicRouteData = this.namesrvController.getRouteInfoManager().pickupTopicRouteData(requestHeader.getTopic());
            //如果找到主题对应的路由信息
            if (topicRouteData != null) {
        
               //该主题为顺序消息
                if (this.namesrvController.getNamesrvConfig().isOrderMessageEnable()) {
        
                    //从NameServer  KVconfig 中获取关于顺序消息相关的配置填充路由信息。
                    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;
            }
    
            //如果找不到路由信息CODE 则使用TOPIC NOT_EXISTS ，表示没有找到对应的路由。
            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;
        }

public TopicRouteData pickupTopicRouteData(final String topic) {
        
            TopicRouteData topicRouteData = new TopicRouteData();
            boolean foundQueueData = false;
            boolean foundBrokerData = false;
            //临时存topic队列所有brokerName
            Set<String> brokerNameSet = new HashSet<String>();
            //设置broker数据信息
            List<BrokerData> brokerDataList = new LinkedList<BrokerData>();
            topicRouteData.setBrokerDatas(brokerDataList);
    
            //设置过滤信息
            HashMap<String, List<String>> filterServerMap = new HashMap<String, List<String>>();
            topicRouteData.setFilterServerTable(filterServerMap);
    
            try {
        
                try {
        
                    this.lock.readLock().lockInterruptibly();
                    //设置Topic消息队列路由信息
                    List<QueueData> queueDataList = this.topicQueueTable.get(topic);
                    if (queueDataList != null) {
        
                        topicRouteData.setQueueDatas(queueDataList);
                        foundQueueData = true;
    
                        //记录topic队列所有brokerName
                        Iterator<QueueData> it = queueDataList.iterator();
                        while (it.hasNext()) {
        
                            QueueData qd = it.next();
                            brokerNameSet.add(qd.getBrokerName());
                        }
    
                        for (String brokerName : brokerNameSet) {
        
                            BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                            if (null != brokerData) {
        
                                //复制一个broker数据信息副本,并填充到brokerDataList中
                                BrokerData brokerDataClone = new BrokerData(brokerData.getCluster(), brokerData.getBrokerName(), (HashMap<Long, String>) brokerData
                                    .getBrokerAddrs().clone());
                                brokerDataList.add(brokerDataClone);
                                foundBrokerData = true;
                                //填充过滤信息
                                for (final String brokerAddr : brokerDataClone.getBrokerAddrs().values()) {
        
                                    List<String> filterServerList = this.filterServerTable.get(brokerAddr);
                                    filterServerMap.put(brokerAddr, filterServerList);
                                }
                            }
                        }
                    }
                } finally {
        
                    this.lock.readLock().unlock();
                }
            } catch (Exception e) {
        
                log.error("pickupTopicRouteData Exception", e);
            }
    
            log.debug("pickupTopicRouteData {} {}", topic, topicRouteData);
    
            //获取成功则正常返回
            if (foundBrokerData && foundQueueData) {
        
                return topicRouteData;
            }
    
            return null;
        }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/93e9fbb4fc75419abc7bd3e45632595b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/f4ea80f16e6c4ed99637476e253ff47d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/f0ffa31fdec640cbab9e45e645fb24aa.png
[20200826104118670.png_pic_center]: /images/20221123/53d6c19696bd4158b99d87cce6dd5d16.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L051YW5fRmVuZw_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/0922c28ac7964a3092c2532c11683643.png
[2020082611573015.png_pic_center]: /images/20221123/db09b3f2d82e4a769693182da36a3546.png
[20200826110914507.png_pic_center]: /images/20221123/57d4247b98004814b7aa7c212b434bd4.png
[20200826121458335.png_pic_center]: /images/20221123/448fe39ae0d947ef85f88c3f480c3c95.png