消息中间件 RocketMQ 高级功能和源码分析（四）

一、 消息中间件 RocketMQ 源码分析：回顾 NameServer 架构设计。

1、RocketMQ 架构设计

消息中间件的设计思路一般是基于主题订阅发布的机制，消息生产者（Producer）发送某一个主题到消息服务器，消息服务器负责将消息持久化存储，消息消费者（Consumer）订阅该兴趣的主题，消息服务器根据订阅信息（路由信息）将消息推送到消费者（Push模式）或者消费者主动向消息服务器拉去（Pull模式），从而实现消息生产者与消息消费者解耦。为了避免消息服务器的单点故障导致的整个系统瘫痪，通常会部署多台消息服务器共同承担消息的存储。那消息生产者如何知道消息要发送到哪台消息服务器呢？如果某一台消息服务器宕机了，那么消息生产者如何在不重启服务情况下感知呢？

NameServer 就是为了解决以上问题设计的。

2、Broker 消息服务器

Broker 消息服务器在启动的时向所有 NameServer注册，消息生产者（Producer）在发送消息时之前先从 NameServer 获取 Broker 服务器地址列表，然后根据负载均衡算法从列表中选择一台服务器进行发送。NameServer与每台Broker保持长连接，并间隔30S检测 Broker 是否存活，如果检测到 Broker 宕机，则从路由注册表中删除。但是路由变化不会马上通知消息生产者。这样设计的目的是为了降低 NameServer 实现的复杂度，在消息发送端提供容错机制保证消息发送的可用性。

3、NameServer 的高可用

NameServer 本身的高可用是通过部署多台 NameServer 来实现，但彼此之间不通讯，也就是 NameServer 服务器之间在某一个时刻的数据并不完全相同，但这对消息发送并不会造成任何影响，这也是 NameServer 设计的一个亮点，总之，RocketMQ 设计追求简单高效。

二、 消息中间件 RocketMQ 源码分析：NameServer 启动步骤一。

1、RocketMQ 源码分析： NameServer 启动流程

步骤一：

解析配置文件，填充 NameServerConfig、NettyServerConfig 属性值，并创建 NamesrvController。

步骤二

根据启动属性创建NamesrvController实例，并初始化该实例。NameServerController实例为NameServer核心控制器。

步骤三

在JVM进程关闭之前，先将线程池关闭，及时释放资源。

2、RocketMQ 源码 启动类：org.apache.rocketmq.namesrv.NamesrvStartup

步骤一：

解析配置文件，填充 NameServerConfig、NettyServerConfig 属性值，并创建 NamesrvController。

代码：NamesrvController#createNamesrvController

//创建NamesrvConfig
final NamesrvConfig namesrvConfig = new NamesrvConfig();
//创建NettyServerConfig
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
//设置启动端口号
nettyServerConfig.setListenPort(9876);
//解析启动-c参数
if (commandLine.hasOption('c')) {String file = commandLine.getOptionValue('c');if (file != null) {InputStream in = new BufferedInputStream(new FileInputStream(file));properties = new Properties();properties.load(in);MixAll.properties2Object(properties, namesrvConfig);MixAll.properties2Object(properties, nettyServerConfig);namesrvConfig.setConfigStorePath(file);System.out.printf("load config properties file OK, %s%n", file);in.close();}
}
//解析启动-p参数
if (commandLine.hasOption('p')) {InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);MixAll.printObjectProperties(console, namesrvConfig);MixAll.printObjectProperties(console, nettyServerConfig);System.exit(0);
}
//将启动参数填充到namesrvConfig,nettyServerConfig
MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);//创建NameServerController
final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);

2、 NamesrvConfig 属性

private String rocketmqHome = System.getProperty(MixAll.ROCKETMQ_HOME_PROPERTY, System.getenv(MixAll.ROCKETMQ_HOME_ENV));
private String kvConfigPath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "kvConfig.json";
private String configStorePath = System.getProperty("user.home") + File.separator + "namesrv" + File.separator + "namesrv.properties";
private String productEnvName = "center";
private boolean clusterTest = false;
private boolean orderMessageEnable = false;

3、 rocketmqHome： rocketmq 主目录

kvConfig： NameServer 存储KV配置属性的持久化路径

configStorePath： nameServer 默认配置文件路径

orderMessageEnable： 是否支持顺序消息

4、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;
private boolean serverPooledByteBufAllocatorEnable = true;
private boolean useEpollNativeSelector = false;

listenPort： NameServer监听端口，该值默认会被初始化为9876
serverWorkerThreads： Netty业务线程池线程个数
serverCallbackExecutorThreads： Netty public任务线程池线程个数，Netty网络设计，根据业务类型会创建不同的线程池，比如处理消息发送、消息消费、心跳检测等。如果该业务类型未注册线程池，则由public线程池执行。
serverSelectorThreads： IO线程池个数，主要是NameServer、Broker端解析请求、返回相应的线程个数，这类线程主要是处理网路请求的，解析请求包，然后转发到各个业务线程池完成具体的操作，然后将结果返回给调用方;
serverOnewaySemaphoreValue： send oneway消息请求并发读（Broker端参数）;
serverAsyncSemaphoreValue： 异步消息发送最大并发度;
serverChannelMaxIdleTimeSeconds ： 网络连接最大的空闲时间，默认120s。
serverSocketSndBufSize： 网络socket发送缓冲区大小。
serverSocketRcvBufSize： 网络接收端缓存区大小。
serverPooledByteBufAllocatorEnable： ByteBuffer是否开启缓存;
useEpollNativeSelector： 是否启用Epoll IO模型。

三、 消息中间件 RocketMQ 源码分析：NameServer 启动步骤二。

1、步骤二

根据启动属性创建 NamesrvController 实例，并初始化该实例。NameServerController 实例为 NameServer 核心控制器

2、代码：NamesrvController#initialize

public boolean initialize() {//加载KV配置this.kvConfigManager.load();//创建NettyServer网络处理对象this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);//开启定时任务:每隔10s扫描一次Broker,移除不活跃的Brokerthis.remotingExecutor =Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));this.registerProcessor();this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {NamesrvController.this.routeInfoManager.scanNotActiveBroker();}}, 5, 10, TimeUnit.SECONDS);//开启定时任务:每隔10min打印一次KV配置this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {NamesrvController.this.kvConfigManager.printAllPeriodically();}}, 1, 10, TimeUnit.MINUTES);return true;
}

四、 消息中间件 RocketMQ 源码分析：NameServer 启动步骤三。

1、步骤三

在 JVM 进程关闭之前，先将线程池关闭，及时释放资源。

2、代码：NamesrvStartup#start

//注册JVM钩子函数代码
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {@Overridepublic Void call() throws Exception {//释放资源controller.shutdown();return null;}
}));

# 消息中间件 RocketMQ 高级功能和源码分析（三）