结合别人的文章，做RocketMQ的一点原理分析，结合源码(尽量)----未完待续

Broker

---

与Namesrv的关系

1.从namesrv获取配置信息

 /**
      * BrokerConfig类
      *
      * broker每隔30秒（此时间无法更改）向所有nameserver发送心跳，心跳包含了自身的topic配置信息。
      * 这里的“此时间无法更改”是别人的总结，还没搞懂为啥此时间不可更改，明明原版注释中写的是取值范围在10，000到60，000之间
      *
      * This configurable item defines interval of topics registration of broker to name server. Allowing values are
      * between 10, 000 and 60, 000 milliseconds.
      */
     private int registerNameServerPeriod = 1000 * 30;

2.namesrv检查broker的心跳

         /**
          * NamesrvController类
          * Namesrv定时检查Broker心跳
          * 每10秒检查一次，时间戳超过2分钟，则认为Broker失效，从brokerLiveTable中移除
          */
         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {

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

 ...

     /**
      * RouteInfoManager类
      * Namesrv查看不活跃|失效的Broker
      */
     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());
             }
         }
     }

3.Namesrv检查与Broker的长连接

 /**
      * BrokerHousekeepingService类
      * 当NameServer和Broker的长连接断掉以后，onChannelDestroy函数会被调用，把这个Broker的信息清理出去。
      */
     @Override
     public void onChannelClose(String remoteAddr, Channel channel) {
         this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
     }

     /**
      * BrokerHousekeepingService类
      * 当NameServer和Broker的长连接断掉以后，onChannelDestroy函数会被调用，把这个Broker的信息清理出去。
      */
     @Override
     public void onChannelException(String remoteAddr, Channel channel) {
         this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
     }

     /**
      * BrokerHousekeepingService类
      * 当NameServer和Broker的长连接断掉以后，onChannelDestroy函数会被调用，把这个Broker的信息清理出去。
      */
     @Override
     public void onChannelIdle(String remoteAddr, Channel channel) {
         this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
     }

Consumer|Producer

---

与Namesrv的关系

     /**
      * ClientConfig类
      *
      * 默认情况下，消费者每隔30秒从nameserver获取所有topic的最新队列情况，
      * 这意味着某个broker如果宕机，客户端最多要30秒才能感知。
      *
      * 默认情况下，生产者每隔30秒从nameserver获取所有topic的最新队列情况，
      * 这意味着某个broker如果宕机，生产者最多要30秒才能感知，在此期间，发往该broker的消息发送失败。
      *
      * Pulling topic information interval from the named server
      */
     private int pollNameServerInterval = 1000 * 30;

与Broker的关系

1.客户端(生产者|消费者)向Broker发送心跳

     /**
      * ClientConfig类
      *
      * 默认情况下，消费者|生产者 每隔30秒向所有broker发送心跳
      *
      * Heartbeat interval in microseconds with message broker
      */
     private int heartbeatBrokerInterval = 1000 * 30;

2.Broker定期检查客户端心跳检查

     /**
      * ClientHousekeepingService类
      *
      * Broker定时清理不活动的客户端(生产者|消费者)连接
      * 每10秒检查一次，时间戳超过2分钟，则关闭客户端(生产者|消费者)连接
      */
     public void start() {

         this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
             @Override
             public void run() {
                 try {
                     ClientHousekeepingService.this.scanExceptionChannel();
                 } catch (Throwable e) {
                     log.error("Error occurred when scan not active client channels.", e);
                 }
             }
         }, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);
     }

 ...

     /**
      * ClientHousekeepingService类
      *
      * Broker定时清理不活动的客户端(生产者|消费者)连接
      * 每10秒检查一次，时间戳超过2分钟，则关闭客户端(生产者|消费者)连接
      */
     private void scanExceptionChannel() {
         this.brokerController.getProducerManager().scanNotActiveChannel();
         this.brokerController.getConsumerManager().scanNotActiveChannel();
         this.brokerController.getFilterServerManager().scanNotActiveChannel();
     }

3.Broker检查与客户端的长连接

     /**
      * ClientHousekeepingService类
      *
      * 当客户端(Producer|Consumer)和Broker的长连接断掉以后，doChannelCloseEvent函数会被调用，把对应客户端(Producer|Consumer)断开连接
      */
     @Override
     public void onChannelClose(String remoteAddr, Channel channel) {
         this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
         this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
         this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
     }

     /**
      * ClientHousekeepingService类
      *
      * 当客户端(Producer|Consumer)和Broker的长连接断掉以后，doChannelCloseEvent函数会被调用，把对应客户端(Producer|Consumer)断开连接
      */
     @Override
     public void onChannelException(String remoteAddr, Channel channel) {
         this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
         this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
         this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
     }

     /**
      * ClientHousekeepingService类
      *
      * 当客户端(Producer|Consumer)和Broker的长连接断掉以后，doChannelCloseEvent函数会被调用，把对应客户端(Producer|Consumer)断开连接
      */
     @Override
     public void onChannelIdle(String remoteAddr, Channel channel) {
         this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
         this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
         this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
     }