美丽乡村网站建设模板,跨境电商工具类产品的网站,wordpress如何建立网站,如何选择网站目标关键词1、RocketMQ组件概述 NameServer NameServer相当于配置中心#xff0c;维护Broker集群、Broker信息、Broker存活信息、主题与队列信息等。NameServer彼此之间不通信#xff0c;每个Broker与集群内所有的Nameserver保持长连接。
2、源码分析NameServer
本文不对 NameServer 与…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 HashMapString/* topic */, ListQueueData topicQueueTable; //3private final HashMapString/* brokerName */, BrokerData brokerAddrTable; //4private final HashMapString/* clusterName */, SetString/* brokerName */ clusterAddrTable; //5private final HashMapString/* brokerAddr */, BrokerLiveInfo brokerLiveTable; //6
代码1NameServer 与 Broker 空闲时长默认2分钟在2分钟内 Nameserver 没有收到 Broker 的心跳包则关闭该连接。
代码2读写锁用来保护非线程安全容器 HashMap。
代码3topicQueueTable主题与队列关系记录一个主题的队列分布在哪些Broker上每个Broker上存在该主题的队列个数。QueueData队列描述信息对应如下属性
private String brokerName; // broker的名称private int readQueueNums; // 读队列个数private int writeQueueNums; // 写队列个数private int perm; // 权限操作代码4brokerAddrTable,所有 Broker 信息使用 brokerName 当key, BrokerData 信息描述每一个 broker 信息。
// broker所属集群
private String cluster; // broker name
private String brokerName;broker 对应的IP:Port,brokerId0表示Master,大于0表示Slave。
代码5clusterAddrTablebroker 集群信息每个集群包含哪些 Broker。
代码6brokerLiveTable当前存活的 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。
