NameServer是整个RMQ的大脑,用于服务发现和服务治理,提供服务调用的解析服务,引导消费者找到对应的服务提供者,那么RMQ的NameServer存储的数据结构是什么?以及自身的高可用如何解决的
整体架构的设计
对于消息中间件,消息中间件根据消息订阅路由将消息推送到消费者,或者消费者主动拉取消息的方式,完成消息生产者和消费者的解耦
为了避免消息服务器的单点故障导致整个系统瘫痪,通常由多个消息服务器共同提供消息存储的功能,对于集群的生存状态的维护,就是NameServer必做的工作
整体的架构如下
Broker作为整个Data信息的存储者,在启动后,向所有NameServer注册
然后生产者在发送消息之前先从NameServer中获取所有的Broker,然后根据负载算法来选择一个发送消息
NameServer负责维护Broker的状态,利用长连接和心跳来进行检测保持
而NameServer的高可用利用了部署多台NameServer来进行维持,但是彼此不通信,保证了AP性质
上面的整体的思考,一下子就考虑到了InfluxDB的数据设计
分为了Meta节点和Data节点,并分别为其设置了CP和AP的特性
接下来,将会是本章的源码部分,我们将利用启动流程来进行相关的源码梳理
启动类为NamesrvStartup
在其中,我们看如下的代码
我们在创建实际的Controller之中
final NamesrvConfig namesrvConfig = new NamesrvConfig();
final NettyServerConfig nettyServerConfig = new NettyServerConfig();
nettyServerConfig.setListenPort(9876);
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();
}
}
if (commandLine.hasOption(‘p’)) {
InternalLogger console = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_CONSOLE_NAME);
MixAll.printObjectProperties(console, namesrvConfig);
MixAll.printObjectProperties(console, nettyServerConfig);
System.exit(0);
}
MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
在整体的填充中
分别先利用-c configFile来利用文件填充
然后是环境变量
之后我们填充完config之后
查看config中的属性值
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;
上面的System.getProperty来获取属性,利用了类似map中的putIfAbsent的api
其中的属性为
rocketmqHome:主目录
kvConfigPath:存储KV配置的路径
configStorePath:利用-c配置的
orderMessageEnable:开启顺序消息
在启动类中,我们还设置了Netty相关的配置
名为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:监听的端口
serverWorkerThreads:工作线程数
serverCallbackExecutorThreads:公共线程池线程数,供消息发送,消息消费,等任务进行使用
serverSelectorThreads :生成EventLoopGroup的时候使用的
serverOnewaySemaphoreValue:消息请求并发数
serverAsyncSemaphoreValue:异步消息发送最大并发数
serverChannelMaxIdleTimeSeconds:网络连接最大空闲时间
serverSocketSndBufSize:发送区大小,默认64k
serverSocketRcvBufSize:接收区大小,默认64k
serverPooledByteBufAllocatorEnable:是否开启缓存
useEpollNativeSelector:是否使用Epoll IO
在之后,我们进行NameSrvController的start函数的调用
其中调用了
controller.initialize();
在initialize的代码中
我们开启了对应的NettyRemotingServer
然后开启了两个定时任务
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.routeInfoManager.scanNotActiveBroker();
}
}, 5, 10, TimeUnit.SECONDS);
这是关于扫描Broker,获取心跳信息的
其中涉及到的Broker状态变更,我们下节再说
然后是第二个定时任务
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.kvConfigManager.printAllPeriodically();
}
}, 1, 10, TimeUnit.MINUTES);
if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {
// Register a listener to reload SslContext
try {
fileWatchService = new FileWatchService(
new String[] {
TlsSystemConfig.tlsServerCertPath,
TlsSystemConfig.tlsServerKeyPath,
TlsSystemConfig.tlsServerTrustCertPath
},
new FileWatchService.Listener() {
boolean certChanged, keyChanged = false;
@Override
public void onChanged(String path) {
if (path.equals(TlsSystemConfig.tlsServerTrustCertPath)) {
log.info(“The trust certificate changed, reload the ssl context”);
reloadServerSslContext();
}
if (path.equals(TlsSystemConfig.tlsServerCertPath)) {
certChanged = true;
}
if (path.equals(TlsSystemConfig.tlsServerKeyPath)) {
keyChanged = true;
}
if (certChanged && keyChanged) {
log.info(“The certificate and private key changed, reload the ssl context”);
certChanged = keyChanged = false;
reloadServerSslContext();
}
}
private void reloadServerSslContext() {
((NettyRemotingServer) remotingServer).loadSslContext();
}
});
} catch (Exception e) {
log.warn(“FileWatchService created error, can’t load the certificate dynamically”);
}
}
每过10分钟,打印全体的KV信息,具体的KV信息接下来再议
在NameServer的start函数中
我们还注册了JVM的钩子函数,这个的确少见,不过这种释放资源的方式的确很好
是一种合理的资源释放手段