//Topic列表信息，保存Topic对应的每个Broker名称，QueueData还包含了mq的读写数量
    private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    //Broker地址信息，保存Broker名称对应的Broker地址信息（就是一台Broker机器）
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    //Broker集群信息，保存Broker集群对应的所有Broker名称
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    //Broker更新信息，保存每台Broker机器的状态信息，NameServer会定期检查该状态
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    //Broker过滤信息，Broker服务端过滤
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

package com.alibaba.rocketmq.namesrv;

public class NamesrvStartup {
  
  /**
   * 程序或命令行启动，调用的入口
   * @param args
   */
  public static void main(String[] args) {
    main0(args);
  }

  public static NamesrcController main0(String[] args) {
    //...
    
    //创建NamesrvConfig配置文件
    final NamesrvConfig namesrvConfig = new NamesrvConfig();
    //创建NettyServerConfig配置文件
    final NettyServerConfig nettyServerConfig = new NettyServerConfig();
    nettyServerConfig.setListenPort(9876);//设置Netty服务端监听端口，默认9876
    
    //创建NamesrvController，传入配置文件构造controller
    final NamesrvController controller = new NamesrvController(namesrcConfig, nettyServerConfig);
    controller.initialize();//初始化controller
    
    //...
  }
  
}

package com.alibaba.rocketmq.namesrv;

public class NamesrvController {
  //Name Server 配置文件
  private final NamesrvCofnig namesrvConfig;
  //Netty 服务端配置文件
  private final NettyServerConfig nettyServerConfig;
  //Netty 服务端
  private RemotingServer remotingServer;
  //Broker状态监听处理程序
  private BrokerHouseKeepingServer brokerHouseKeepingService;
  //线程池 用于处理消息请求DefaultRequestProcessor
  private ExecutorService remotingService;
  
  //...
  
  /**
     * 实例化NamesrvController
     * @param namesrvConfig Name Server 配置文件
     * @param nettyServerConfig Netty 服务端配置文件
     */
  public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {
    this.namesrvConfig = namesrvConfig;
    this.nettyServerConfig = nettyServerConfig;
    //...
    //保存所有的路由信息
    this.routeInfoManager = new RouteInfoManager();
    //Broker状态事件处理程序，处理Broker的异常、关闭等状态
    this.brokerHousekeepingService = new BrokerHousekeepingService(this);
  }
  
  /**
   * 初始化
   * @return
   */
  public boolean initialize() {
    //...
    
    //创建Netty服务端，nettyServerConfig为Netty服务端相关配置文件，例如前面在NamesrvStartup中配置了监听端口9876
    //brokerHouseKeepingServer broker状态处理程序
    this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
    //创建一个固定大小的线程池，根据nettyServerConfig配置文件提供的默认工作线程数，默认值为8
    this.remotingExecutor = Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
    //注册请求处理程序，当Netty服务端接收到客户端的请求时，会调用该处理程序
    this.registerProcessor();
    
    //...
  }

  /**
  	* 给Netty服务端注册请求处理程序，使用线程池来处理请求
    */
  private void registerProcessor() {
  	//...

  	this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingServer);
  }
}

package com.alibaba.rocketmq.namesrv.routeinfo;

public class RouteInfoManager {

    // Broker上报注册时会更新一下5个集合

    // 使用读写锁控制broker注册并发
    private final ReadWriteLock lock = new ReentrantReadWriteLock();

	  //Topic列表信息，保存Topic对应的每个Broker名称，QueueData还包含了mq的读写数量
    private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
    //Broker地址信息，保存Broker名称对应的Broker地址信息（就是一台Broker机器）
    private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
    //Broker集群信息，保存Broker集群对应的所有Broker名称
    private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
    //Broker更新信息，保存每台Broker机器的状态信息，NameServer会定期检查该状态
    private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
    //Broker过滤信息，Broker服务端过滤
    private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
    
    //...
    
    /**
      * 注册新的Broker，将新的Broker添加到brokderAddrTable等信息中
      */
    public RegisterBrokerResult registerBroker(...) {
      //...
      //更新Broker集群信息
      this.clusterAddrTable.put(clusterName, brokerNames);
      //更新Broker地址信息
      this.brokerAddrTable.put(brokerName, brokerData);
      //更新Broker状态信息
      this.brokerLiveTable.put(...);
      //更新Broker过滤信息
      this.filterServerTable.put(brokerAddr, filterServerList);
      // 更新Topic信息
      this.topicQueueTable.put(topicConfig.getTopicName(), queueDataList);
      //...
    }
}

package com.alibaba.rocketmq.remoting.netty;

/**
 * Netty服务端公共抽象类
 */
public abstract class NettyRemotingAbstract {

  //...

  /**
   * 新建线程监听事件队列
   */
  class NettyEventExecuter extends ServiceThread {
  
    //事件队列，这里使用LinkedBlockingQueue队列，基于链表的阻塞队列，是线程安全的队列。新事件自动加入到队尾，
    private final LinkedBlockingQueue<NettyEvent> eventQueue = new LinkedBlockingQueue<NettyEvent>();
    //...
    
    /**
     * 添加新事件
     * @param event
     */
    public void putNettyEvent(final NettyEvent event) {
      this.eventQueue.add(event);
    }
    
    @Override
    public void run() {
      //请求事件处理程序 BrokerHouseKeepingService
      final ChannelEventListener listener = NettyRemotingAbstract.this.getChannelEventListener();
      
      while(!this.isStoped()) {
        //每3秒访问一次事件队列
        NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS);
        if (event != null && listener != null) {
          //判断事件类型
          switch (event.getType()) {
          case IDLE://失效
            //调用 BrokerHouseKeepingService 的 onChannelIdle
            listener.onChannelIdle(event.getRemoteAddr(), event.getChannel());
            break;
          case CLOSE://关闭
            //调用 BrokerHouseKeepingService 的 onChannelClose
            listener.onChannelClose(event.getRemoteAddr(), event.getChannel());
            break;
          case CONNECT://连接
            //调用 BrokerHouseKeepingService 的 onChannelConnect
            listener.onChannelConnect(event.getRemoteAddr(), event.getChannel());
            break;
          case EXCEPTION://异常
            //调用 BrokerHouseKeepingService 的 onChannelException
            listener.onChannelException(event.getRemoteAddr(), event.getChannel());
            break;
          default:
              break;
          }
        }
      }
    }
  }
  
}

  package com.alibaba.rocketmq.namesrv.routeinfo;

  /**
   * Broker状态事件处理程序，处理Broker状态的服务，实现了ChannelEventListener接口。
   * 当Netty有新事件时，将调用ChannelEventListener接口处理事件。
   */
  public class BrokerHouseKeepingService implements ChannelEventListener {
  
    //...
    
    @Override
    public void onChannelClose(String remoteAddr, Channel channel) {
      //管道关闭时，将broker从RouteInfoManager中移除
      this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
    }
    
    @Override
    public void onChannelException(String remoteAddr, Channel channel) {
      //管道异常时，将broker从RouteInfoManager中移除
      this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
    }

    @Override
    public void onChannelIdle(String remoteAddr, Channel channel) {
      //管道失效时，将broker从RouteInfoManager中移除
      this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
    }
    
    //...
  }

package com.alibaba.rocketmq.namesrv.routeinfo;

public class RouteInfoManager {

	//...
    
    /**
     * 销毁管道
     * @param remoteAddr 客户端地址信息
     * @param channel 客户端管道
     */
    public void onChannelDestroy(String remoteAddr, Channel channel) {
    
      //1.根据channel客户端管道找到brokerAddr对应的broker地址信息
      //2.根据brokerAddr地址将broker从brokerLiveTable和filterServerTable移除
      this.brokerLiveTable.remove(brokerAddrFound);
      this.filterServerTable.remove(brokerAddrFound);
      //3.根据brokerAddr遍历brokerAddrTable，将broker从brokerAddrTable中移除
      this.brokerAddrTable.remove();
      //4.根据brokerAddr遍历clusterAddrTable，将broker从clusterAddrTable中移除
      this.clusterAddrTable.remove();
      //4.根据brokerAddr遍历topicQueueTable，将broker从topicQueueTable中移除
      this.topicQueueTable.remove();
      
    }
}

package com.alibaba.rocketmq.remoting.netty;

/**
 * Netty 服务端
 */
public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {
  
  /**
   * 通过Netty配置文件和管道事件监听器实例化Netty服务端
   * @param nettyServerConfig Netty配置文件
   * @param channelEventListener 管道事件监听器 BrokerHouseKeepingService实现了该接口
   */
  public NettyRemotingServer(final NettyServerConfig nettyServerConfig, final ChannelEventListener channelEventListener) {
    //...
    //Netty 服务端启动辅助类
    this.serverBootstrap = new ServerBootstrap();
    //Netty 配置文件
    this.nettyServerConfig = nettyServerConfig;
    //管道事件监听器 BrokerHouseKeepingService实现了该接口
    this.channelEventListener = channelEventListener;
    //...
  }
  
  @Override
  public void start() {
    //正常的启动Netty
    this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector).channel(NioServerSocketChannel.class)
      .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))//监听配置文件设置的9876端口
      .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
          ch.pipeline().addLast(
            new NettyEncoder(),//自定义Netty编码类
            new NettyDecoder(),//自定义Netty解码类
            new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),//心跳空闲检查时间
            new NettyConnetManageHandler(),//处理管道连接、端口、异常等事件
            new NettyServerHandler());//处理管道消息事件
        }
      });
      
    this.nettyEventExecutor.start();//启动一个线程，监听事件队列。就是上面提到的管道断开、异常或关闭时的事件队列
  }
  
}

package com.alibaba.rocketmq.remoting.netty;

/**
 * Netty 服务端
 */
public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {
  //...
  
  class NettyConnetManageHandler extends ChannelDuplexHandler {
    //...
    
   @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
      //获取客户端远程地址
      final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
      //创建一个NettyEvent事件，标记该事件类型为CLOSE，然后将该事件添加到nettyEventExecuter的eventQueue队列中
      NettyRemotingServer.this.putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress.toString(), ctx.channel()));
  }
}

package com.alibaba.rocketmq.remoting.netty;

/**
 * Netty 服务端
 */
public class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {
  //...
  
  class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {
    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
      //调用NettyRemoteAbstract的processMessageReceived方法来处理请求
      processMessageReceived(ctx, msg);
    }
}

package com.alibaba.rocketmq.remoting.netty;

/**
 * Netty服务端公共抽象类
 * @author shijia.wxr
 */
public abstract class NettyRemotingAbstract {
  //NamesrvController注册的默认处理程序，以及用于执行该处理程序的线程池
  protected Pair<NettyRequestProcessor, ExecutorService> defaultRequestProcessor;
  
  public void processMessageReceived(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
    final RemotingCommand cmd = msg;
    //判断事件类型
    switch (cmd.getType()) {
      case REQUEST_COMMAND://请求事件
          processRequestCommand(ctx, cmd);
          break;
      case RESPONSE_COMMAND://响应事件
          processResponseCommand(ctx, cmd);
          break;
      default:
          break;
    }
  }
  
  /**
   * 请求事件处理程序
   * @param ctx
   * @param cmd
   */
  public void processRequestCommand(final ChannelHandlerContext ctx, final RemotingCommand cmd) {
    final Pair<NettyRequestProcessor, ExecutorService> matched = this.processorTable.get(cmd.getCode());
    //使用默认的请求处理程序
    final Pair<NettyRequestProcessor, ExecutorService> pair = null == matched ? this.defaultRequestProcessor : matched;
    
    //把请求封装成一个Task
    Runnable run = new Runnable() {
      @Override
      public void run() {
        //...
        
        //调用默认的请求处理程序处理请求 DefaultRequestProcessor的processRequest方法
        final RemotingCommand response = pair.getObject1().processRequest(ctx, cmd);
        
        //...
      }
    }
    //将Task任务提交到线程池中执行
    pair.getObject2().submit(run);
  }
}

package com.alibaba.rocketmq.namesrv.processor;

/**
  * 默认请求处理程序
  */
public class DefaultRequestProcessor implements NettyRequestProcessor {
  
  /**
   * 处理请求
   * @param ctx 管道
   * @param request 请求消息
   * @return 响应消息
   * @throws RemotingCommandException
   */
  @Override
  public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    //...
    //根据请求消息类型
    switch (request.getCode()) {
    case RequestCode.REGISTER_BROKER://注册新Broker
      return this.registerBroker(ctx, request);
    case RequestCode.UNREGISTER_BROKER://移除Broker
      return this.unregisterBroker(ctx, request);
    //...
    default:
      break;
    }
    return null;
  }
  
  /**
   * 注册新Broker
   * @param ctx
   * @param request
   * @return
   * @throws RemotingCommandException
   */
  public RemotingCommand registerBroker(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
    //...
    //通过调用RouteInfoManager的registerBroker方法来注册新的Broker
    this.namesrvController.getRouteInfoManager().registerBroker(...);
    //...
  }
}