上海opencart网站建设全网推广外包公司
1、RocketMQ组件概述
- NameServer
NameServer相当于配置中心，维护Broker集群、Broker信息、Broker存活信息、主题与队列信息等。NameServer彼此之间不通信，每个Broker与集群内所有的Nameserver保持长连接。
2、源码分析NameServer
本文不对 NameServer 与 Broker、Producer 集群、Consumer 集群的网络通信做详细解读（该系列后续专门进行讲解）
本文重点关注 NameServer 作为 MQ 集群的配置中心存储什么信息。
2.1 源码分析NamesrvController
NameserController 是 NameServer 模块的核心控制类。
2.1.1 NamesrvConfig
NamesrvConfig,主要指定 nameserver 的相关配置属性：
- kvConfigPath(kvConfig.json)。
- mqhome/namesrv/namesrv.properties。
- orderMessageEnable，是否开启顺序消息功能，默认为false。
2.1.2 ScheduledExecutorService
private final ScheduledExecutorService scheduledExecutorService = Executors.NameServer 定时任务执行线程池，默认定时执行两个任务：
- 任务1、每隔 10s 扫描 broker ,维护当前存活的Broker信息。
- 任务2、每隔 10s 打印KVConfig 信息。
2.1.3 KVConfigManager
读取或变更NameServer的配置属性，加载 NamesrvConfig 中配置的配置文件到内存，此类一个亮点就是使用轻量级的非线程安全容器，再结合读写锁对资源读写进行保护。尽最大程度提高线程的并发度。
2.1.4 RouteInfoManager
NameServer 数据的载体，记录 Broker、Topic 等信息。
private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2; //@1private final ReadWriteLock lock = new ReentrantReadWriteLock(); //@2private final HashMap<String/* topic */, List<QueueData>> topicQueueTable; //@3private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable; //@4private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable; //@5private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable; //@6代码@1，NameServer 与 Broker 空闲时长，默认2分钟，在2分钟内 Nameserver 没有收到 Broker 的心跳包，则关闭该连接。
代码@2，读写锁，用来保护非线程安全容器 HashMap。
代码@3，topicQueueTable，主题与队列关系，记录一个主题的队列分布在哪些Broker上，每个Broker上存在该主题的队列个数。QueueData队列描述信息，对应如下属性：
private String brokerName; // broker的名称private int readQueueNums; // 读队列个数private int writeQueueNums; // 写队列个数private int perm; // 权限操作
代码@4，brokerAddrTable,所有 Broker 信息，使用 brokerName 当key, BrokerData 信息描述每一个 broker 信息。
// broker所属集群
private String cluster; // broker name
private String brokerName;broker 对应的IP:Port,brokerId=0表示Master,大于0表示Slave。 代码@5，clusterAddrTable，broker 集群信息，每个集群包含哪些 Broker。
代码@6，brokerLiveTable，当前存活的 Broker,该信息不是实时的，NameServer 每10S扫描一次所有的 broker,根据心跳包的时间得知 broker的状态，该机制也是导致当一个 Broker 进程假死后，消息生产者无法立即感知，可能继续向其发送消息，导致失败（非高可用），如何保证消息发送高可用，请关关注该系列后续文章。
2.1.5 BrokerHousekeepingService
BrokerHouseKeepingService 实现 ChannelEventListener接口，可以说是通道在发送异常时的回调方法（Nameserver与 Broker的连接通道在关闭、通道发送异常、通道空闲时），在上述数据结构中移除已宕机的 Broker。
public interface ChannelEventListener {void onChannelConnect(final String remoteAddr, final Channel channel);void onChannelClose(final String remoteAddr, final Channel channel);void onChannelException(final String remoteAddr, final Channel channel);void onChannelIdle(final String remoteAddr, final Channel channel);2.1.6 NettyServerConfig、RemotingServer 、ExecutorService
这三个属性与网络通信有关，NameServer 与 Broker、Producer、Consume 之间的网络通信，基于 Netty实现。本文借这个机会再次探究 Netty 线程模型与 Netty实战技巧。
源码分析网络通讯之前，我们关注如下问题：
- NettyServerConfig 的配置含义
- Netty 线程模型中 EventLoopGroup、EventExecutorGroup 之间的区别与作用
- 在 Channel 的整个生命周期中，如何保证 Channel 的读写事件至始至终使用同一个线程处理
首先我们先过一下NettyServerConfig中的配置属性：
private int listenPort = 8888;private int serverWorkerThreads = 8;private int serverCallbackExecutorThreads = 0;private int serverSelectorThreads = 3;private int serverOnewaySemaphoreValue = 256;private int serverAsyncSemaphoreValue = 64;private int serverChannelMaxIdleTimeSeconds = 120;private int serverSocketSndBufSize = NettySystemConfig.socketSndbufSize;private int serverSocketRcvBufSize = NettySystemConfig.socketRcvbufSize;我们带着上面的疑问开始源码分析 org.apache.rocketmq.remoting.netty.NettyRemotingServer。