文章目录
源文档参照
https://blog.csdn.net/lm324114/article/details/106968106
码云代码
https://gitee.com/yanweiling/rocketmq-all-4.3.0
架构设计
NameServer 作用
NameServer主要作用是为消息生产者和消息消费者提供关于主题Topic的路由信息,
那么NameServer需要存储路由的基础信息,还要能够管理Broker节点,包括路由注册、路由删除等功能。
路由元数据结构
NameServer路由实现类: org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager,在了解路由注册之前,我们首先看一下NameServer到底存储哪些信息。
private final HashMap<String/* topic */, List<QueueData>> topicQueueTable;
private final HashMap<String/* brokerName */, BrokerData> brokerAddrTable;
private final HashMap<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;
private final HashMap<String/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;
private final HashMap<String/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;
- topicQueueTable: Topic消息队列路由信息,消息发送时根据路由表进行负载均衡。
- brokerAddrTable: Broker基础信息,包含BrokerName、所属集群名称、主备Broker地址。
- clusterAddrTable: Broker集群信息,存储集群中所有Broker名称。
- brokerLiveTable: Broker状态信息。NameServer每次收到心跳包时会替换该信息。
- filterServerTable: Broker上的FilterServer列表,用于类模式消息过滤
NameServer启动流程
NameServer启动类:
步骤一
首先来解析配置文件,需要填充NamesrvConfig、NettyServerConfig属性值。
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, " + file + "%n");
in.close();
}
}
if (commandLine.hasOption('p')) {
MixAll.printObjectProperties(null, namesrvConfig);
MixAll.printObjectProperties(null, nettyServerConfig);
System.exit(0);
}
MixAll.properties2Object(ServerUtil.commandLine2Properties(commandLine), namesrvConfig);
从代码我们可以知道先创建NamesrvConfig(NameServer业务参数)、NettyServerConfig(NameServer网络参数),然后在解析启动时把指定的配置文件或启动命令中的选项值,填充到namesrvConfig,nettyServerConfig对象。
参数来源有如下两种方式:
1、-c configFile通过-c命令指定配置文件的路径
2、使用"–属性名 属性值",例如–listenPort 9876
NameServerConfig属性
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;
- rocketmqhome: rocketmq主目录,可以通过-Drocketmq.home.dir=path或通过设置环境变量ROCKETMQ_HOME来配置RocketMQ的主目录。
- kvConfigPath: NameServer存储KV配置属性的持久化路径。
- configStorePath: nameServer默认配置文件路径,不生效。nameServer启动时如果要通过配置文件配置NameServer启动属性的话,请使用-c选项。
- orderMessageEnable: 是否支持顺序消息,默认是不支持
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网络设计,根据业务类型会常见不同的线程池,比如处理消息发送、消息消费、心跳检测等等。如果该业务类型(RequestCode)未注册线程池,则由public线程池执行。
-
serverSelectorThreads:IO线程池线程个数,主要是NameServer、Broker端解析请求、返回相应的线程个数,这类线程主要是处理网络请求的,解析请求包,然后转发到各个业务线程池完成具体的业务操作,然后将结果返回调用方。
-
serverOnewaySemaphoreValue: send oneway消息请求并发度(Broker端参数)。
-
serverAsyncSemaphoreValue: 异步消息发送最大并发度(Broker端参数)。
-
serverChannelMaxIdleTimeSeconds: 网络连接最大空闲时间,默认120s。如果连接空闲时间超过该参数设置的值,连接将被关闭。
-
serverSocketSndBufSize: 网络socket发送缓存区大小,默认64k。
-
serverSocketRecBufSize: 网络socket接收缓存区大小,默认64k。
-
serverPooledByteBufAllocatorEnable: ByteBuffer是否开启缓存,建议开启。
-
useEpollNativeSelector: 是否启用Epoll IO模型,Linux环境建议开启。
在启动NameServer时,可以先使用./mqnameserver -c configFile -p 打印当前加载的配置属性
步骤2
根据启动属性创建NamesrvController实例,并初始化该实例,NamesrvController实例为NameServer核心控制器。
NamesrvController#initialize代码片段
public boolean initialize() {
this.kvConfigManager.load();
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
this.remotingExecutor =
Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
this.registerProcessor();
//NameServer每隔10s扫描一次Broker,移除处于不激活状态的Broker。
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
NamesrvController.this.routeInfoManager.scanNotActiveBroker();
}
}, 5, 10, TimeUnit.SECONDS);
// NameServer每隔十分钟打印一次KV配置。
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");
}
}
return true;
}
加载KV配置,创建NettyServer网络处理对象,然后开启两个定时任务,在RocketMQ中此类定时任务统称为心跳检测。
-
定时任务1: NameServer每隔10s扫描一次Broker,移除处于不激活状态的Broker。
-
定时任务2: NameServer每隔十分钟打印一次KV配置。
步骤三
注册JVM钩子函数并启动服务器,以便监听Broker、消息生产者的网络请求
NamesrvStartup#start
public static NamesrvController start(final NamesrvController controller) throws Exception {
//忽略。。。
boolean initResult = controller.initialize();
if (!initResult) {
controller.shutdown();
System.exit(-3);
}
//注册钩子
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {
@Override
public Void call() throws Exception {
controller.shutdown();
return null;
}
}));
controller.start();
return controller;
}
Broker启动
启动流程
路由注册
RocketMQ路由注册是通过Broker与NameServer的心跳功能实现的。Broker启动时向集群中所有的NameServer发送心跳语句,每隔30秒向集群中所有NameServer发送心跳包,NameServer收到Broker心跳包时会更新brokerLiveTable缓存中BrokerLiveInfo的lastUpdateTimestamp,然后NameServer每隔10秒扫描brokerLiveTable,如果连续120s没有收到心跳包,NameServer将移除该Broker的路由信息同时关闭Socket连接。
BrokerController.start
//启动broker时立即向所有的nameserver 注册broker信息，并开启定时任务，定时刷新注册broker
this.registerBrokerAll(true, false, true);
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
//发送心跳包
BrokerController.this.registerBrokerAll(true, false, brokerConfig.isForceRegister());
} catch (Throwable e) {
log.error("registerBrokerAll Exception", e);
}
}
}, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);
注册BrokerOuteAPI#registerBroker(网络发送代码)
public List<RegisterBrokerResult> registerBrokerAll(
final String clusterName,
final String brokerAddr,
final String brokerName,
final long brokerId,
final String haServerAddr,
final TopicConfigSerializeWrapper topicConfigWrapper,
final List<String> filterServerList,
final boolean oneway,
final int timeoutMills,
final boolean compressed) {
final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();
List<String> nameServerAddressList = this.remotingClient.getNameServerAddressList();
if (nameServerAddressList != null && nameServerAddressList.size() > 0) {
//创建请求头
final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
requestHeader.setBrokerAddr(brokerAddr);
requestHeader.setBrokerId(brokerId);//brokerId,0: Master ; 大于0: Slave。
requestHeader.setBrokerName(brokerName);
requestHeader.setClusterName(clusterName);//集群名称
requestHeader.setHaServerAddr(haServerAddr);//haServerAddr: master地址,初次请求时该值为空,slave向Nameserver注册后返回。
requestHeader.setCompressed(compressed);
RegisterBrokerBody requestBody = new RegisterBrokerBody();
/**
* 主题配置,topicConfigWrapper内部封装的是TopicConfigManager中的topicConfigTable,
* 内部存储的是Broker启动时默认的一些Topic,MixAll.SELF_TEST_TOPIC、
* MixAll.DEFAULT_TOPICC(AutoCreateTopicEnable=true)、
* MixAll.BENCHMARK_TOPIC、MixAll.OFFSET_MOVED_EVENT、
* BrokerConfig#brokerClusterName、BrokerConfig#brokerName。
* Broker中Topic默认存储在${Rocket_Home}/store/config/topics.json中。
*/
requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
requestBody.setFilterServerList(filterServerList);//消息过滤服务器列表。
final byte[] body = requestBody.encode(compressed);
final int bodyCrc32 = UtilAll.crc32(body);
requestHeader.setBodyCrc32(bodyCrc32);
final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
for (final String namesrvAddr : nameServerAddressList) {
brokerOuterExecutor.execute(new Runnable() {
@Override
public void run() {
try {
//分别向NameServer注册
RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
if (result != null) {
registerBrokerResultList.add(result);
}
log.info("register broker to name server {} OK", namesrvAddr);
} catch (Exception e) {
log.warn("registerBroker Exception, {}", namesrvAddr, e);
} finally {
countDownLatch.countDown();
}
}
});
}
try {
countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
}
}
return registerBrokerResultList;
}
NameServer如何处理心跳
public static NamesrvController start(final NamesrvController controller) throws Exception {
boolean initResult = controller.initialize();
//调用jvm停止钩子
controller.start();//
return controller;
}
NameServer启动的时候，有两个主要方法
1.初始化—> controller.initialize()—>注册了默认请求处理器
2.start方法–> this.remotingServer.start();—>启动netty服务
public void start() {
//忽略
ServerBootstrap childHandler =
this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
.channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
.option(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.SO_KEEPALIVE, false)
.childOption(ChannelOption.TCP_NODELAY, true)
.childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
.childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
.localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort()))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline()
.addLast(defaultEventExecutorGroup, HANDSHAKE_HANDLER_NAME,
new HandshakeHandler(TlsSystemConfig.tlsMode))
.addLast(defaultEventExecutorGroup,
new NettyEncoder(),//编码器
new NettyDecoder(),//解码器
new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),//空闲检测
new NettyConnectManageHandler(),
new NettyServerHandler()//请求handler
);
}
});
//启动netty服务
ChannelFuture sync = this.serverBootstrap.bind().sync();
InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
this.port = addr.getPort();
当有broker的socket连接请求到该nettfy服务上以后，请求会通过handler链条，最终进入到NettyServerHandler的逻辑处理模块中
===>
class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, RemotingCommand msg) throws Exception {
processMessageReceived(ctx, msg);
}
}
对于broker的心跳，属于request请求，所以进入到NettyRemotingAbstract.processRequestCommand的方法中
请求收到后，进入业务处理模块；
业务处理采用异步的，不同的请求处理，采用对应的线程池来执行
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;
//忽略
final RemotingCommand response = pair.getObject1().processRequest(ctx, cmd);--->继而调用DefaultRequestProcessor的processRequest方法
//忽略
—>继而调用DefaultRequestProcessor的processRequest方法
@Override
public RemotingCommand processRequest(ChannelHandlerContext ctx,
RemotingCommand request) throws RemotingCommandException {
switch (request.getCode()) {
//忽略
case RequestCode.REGISTER_BROKER:
Version brokerVersion = MQVersion.value2Version(request.getVersion());
if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()) {
return this.registerBrokerWithFilterServer(ctx, request);
} else {
return this.registerBroker(ctx, request);
}
//忽略
发送消息
生产者启动
Broker启动思维导图
https://rocketmq.apache.org/docs/motivation/
