# 消息中间件 RocketMQ 高级功能和源码分析(四)
消息中间件 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、NameServer 启动流程图:
3、RocketMQ 源码 启动类:org.apache.rocketmq.namesrv.NamesrvStartup.java
源码:
/*
D:\RocketMQ\rocketmq-master\namesrv\src\main\java\org\apache\rocketmq\namesrv\NamesrvStartup.java* Licensed to the Apache Software Foundation (ASF) under one or more* contributor license agreements. See the NOTICE file distributed with* this work for additional information regarding copyright ownership.* The ASF licenses this file to You under the Apache License, Version 2.0* (the "License"); you may not use this file except in compliance with* the License. You may obtain a copy of the License at** http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/
package org.apache.rocketmq.namesrv;import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.joran.JoranConfigurator;
import ch.qos.logback.core.joran.spi.JoranException;
import java.io.BufferedInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.concurrent.Callable;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.PosixParser;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.srvutil.ServerUtil;
import org.apache.rocketmq.srvutil.ShutdownHookThread;
import org.slf4j.LoggerFactory;public class NamesrvStartup {private static InternalLogger log;private static Properties properties = null;private static CommandLine commandLine = null;public static void main(String[] args) {main0(args);}public static NamesrvController main0(String[] args) {try {NamesrvController controller = createNamesrvController(args);start(controller);String tip = "The Name Server boot success. serializeType=" + RemotingCommand.getSerializeTypeConfigInThisServer();log.info(tip);System.out.printf("%s%n", tip);return controller;} catch (Throwable e) {e.printStackTrace();System.exit(-1);}return null;}public static NamesrvController createNamesrvController(String[] args) throws IOException, JoranException {System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, Integer.toString(MQVersion.CURRENT_VERSION));//PackageConflictDetect.detectFastjson();Options options = ServerUtil.buildCommandlineOptions(new Options());commandLine = ServerUtil.parseCmdLine("mqnamesrv", args, buildCommandlineOptions(options), new PosixParser());if (null == commandLine) {System.exit(-1);return null;}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);if (null == namesrvConfig.getRocketmqHome()) {System.out.printf("Please set the %s variable in your environment to match the location of the RocketMQ installation%n", MixAll.ROCKETMQ_HOME_ENV);System.exit(-2);}LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();JoranConfigurator configurator = new JoranConfigurator();configurator.setContext(lc);lc.reset();configurator.doConfigure(namesrvConfig.getRocketmqHome() + "/conf/logback_namesrv.xml");log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);MixAll.printObjectProperties(log, namesrvConfig);MixAll.printObjectProperties(log, nettyServerConfig);final NamesrvController controller = new NamesrvController(namesrvConfig, nettyServerConfig);// remember all configs to prevent discardcontroller.getConfiguration().registerConfig(properties);return controller;}public static NamesrvController start(final NamesrvController controller) throws Exception {if (null == controller) {throw new IllegalArgumentException("NamesrvController is null");}boolean initResult = controller.initialize();if (!initResult) {controller.shutdown();System.exit(-3);}Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {@Overridepublic Void call() throws Exception {controller.shutdown();return null;}}));controller.start();return controller;}public static void shutdown(final NamesrvController controller) {controller.shutdown();}public static Options buildCommandlineOptions(final Options options) {Option opt = new Option("c", "configFile", true, "Name server config properties file");opt.setRequired(false);options.addOption(opt);opt = new Option("p", "printConfigItem", false, "Print all config item");opt.setRequired(false);options.addOption(opt);return options;}public static Properties getProperties() {return properties;}
}
4、org.apache.rocketmq.namesrv.NamesrvController.java
源码:
/*
D:\RocketMQ\rocketmq-master\namesrv\src\main\java\org\apache\rocketmq\namesrv\NamesrvController.java* Licensed to the Apache Software Foundation (ASF) under one or more* contributor license agreements. See the NOTICE file distributed with* this work for additional information regarding copyright ownership.* The ASF licenses this file to You under the Apache License, Version 2.0* (the "License"); you may not use this file except in compliance with* the License. You may obtain a copy of the License at** http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/
package org.apache.rocketmq.namesrv;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.apache.rocketmq.common.Configuration;
import org.apache.rocketmq.common.ThreadFactoryImpl;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.namesrv.kvconfig.KVConfigManager;
import org.apache.rocketmq.namesrv.processor.ClusterTestRequestProcessor;
import org.apache.rocketmq.namesrv.processor.DefaultRequestProcessor;
import org.apache.rocketmq.namesrv.routeinfo.BrokerHousekeepingService;
import org.apache.rocketmq.namesrv.routeinfo.RouteInfoManager;
import org.apache.rocketmq.remoting.RemotingServer;
import org.apache.rocketmq.remoting.common.TlsMode;
import org.apache.rocketmq.remoting.netty.NettyRemotingServer;
import org.apache.rocketmq.remoting.netty.NettyServerConfig;
import org.apache.rocketmq.remoting.netty.TlsSystemConfig;
import org.apache.rocketmq.srvutil.FileWatchService;public class NamesrvController {private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);private final NamesrvConfig namesrvConfig;private final NettyServerConfig nettyServerConfig;private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(new ThreadFactoryImpl("NSScheduledThread"));private final KVConfigManager kvConfigManager;private final RouteInfoManager routeInfoManager;private RemotingServer remotingServer;private BrokerHousekeepingService brokerHousekeepingService;private ExecutorService remotingExecutor;private Configuration configuration;private FileWatchService fileWatchService;public NamesrvController(NamesrvConfig namesrvConfig, NettyServerConfig nettyServerConfig) {this.namesrvConfig = namesrvConfig;this.nettyServerConfig = nettyServerConfig;this.kvConfigManager = new KVConfigManager(this);this.routeInfoManager = new RouteInfoManager();this.brokerHousekeepingService = new BrokerHousekeepingService(this);this.configuration = new Configuration(log,this.namesrvConfig, this.nettyServerConfig);this.configuration.setStorePathFromConfig(this.namesrvConfig, "configStorePath");}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();this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {NamesrvController.this.routeInfoManager.scanNotActiveBroker();}}, 5, 10, TimeUnit.SECONDS);this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {@Overridepublic void run() {NamesrvController.this.kvConfigManager.printAllPeriodically();}}, 1, 10, TimeUnit.MINUTES);if (TlsSystemConfig.tlsMode != TlsMode.DISABLED) {// Register a listener to reload SslContexttry {fileWatchService = new FileWatchService(new String[] {TlsSystemConfig.tlsServerCertPath,TlsSystemConfig.tlsServerKeyPath,TlsSystemConfig.tlsServerTrustCertPath},new FileWatchService.Listener() {boolean certChanged, keyChanged = false;@Overridepublic 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;}private void registerProcessor() {if (namesrvConfig.isClusterTest()) {this.remotingServer.registerDefaultProcessor(new ClusterTestRequestProcessor(this, namesrvConfig.getProductEnvName()),this.remotingExecutor);} else {this.remotingServer.registerDefaultProcessor(new DefaultRequestProcessor(this), this.remotingExecutor);}}public void start() throws Exception {this.remotingServer.start();if (this.fileWatchService != null) {this.fileWatchService.start();}}public void shutdown() {this.remotingServer.shutdown();this.remotingExecutor.shutdown();this.scheduledExecutorService.shutdown();if (this.fileWatchService != null) {this.fileWatchService.shutdown();}}public NamesrvConfig getNamesrvConfig() {return namesrvConfig;}public NettyServerConfig getNettyServerConfig() {return nettyServerConfig;}public KVConfigManager getKvConfigManager() {return kvConfigManager;}public RouteInfoManager getRouteInfoManager() {return routeInfoManager;}public RemotingServer getRemotingServer() {return remotingServer;}public void setRemotingServer(RemotingServer remotingServer) {this.remotingServer = remotingServer;}public Configuration getConfiguration() {return configuration;}
}
5、org.apache.rocketmq.common.namesrv.NamesrvConfig.java
源码:
/*
D:\RocketMQ\rocketmq-master\common\src\main\java\org\apache\rocketmq\common\namesrv\NamesrvConfig.java* Licensed to the Apache Software Foundation (ASF) under one or more* contributor license agreements. See the NOTICE file distributed with* this work for additional information regarding copyright ownership.* The ASF licenses this file to You under the Apache License, Version 2.0* (the "License"); you may not use this file except in compliance with* the License. You may obtain a copy of the License at** http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*//*** $Id: NamesrvConfig.java 1839 2013-05-16 02:12:02Z vintagewang@apache.org $*/
package org.apache.rocketmq.common.namesrv;import java.io.File;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.InternalLogger;
import org.apache.rocketmq.logging.InternalLoggerFactory;public class NamesrvConfig {private static final InternalLogger log = InternalLoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);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;public boolean isOrderMessageEnable() {return orderMessageEnable;}public void setOrderMessageEnable(boolean orderMessageEnable) {this.orderMessageEnable = orderMessageEnable;}public String getRocketmqHome() {return rocketmqHome;}public void setRocketmqHome(String rocketmqHome) {this.rocketmqHome = rocketmqHome;}public String getKvConfigPath() {return kvConfigPath;}public void setKvConfigPath(String kvConfigPath) {this.kvConfigPath = kvConfigPath;}public String getProductEnvName() {return productEnvName;}public void setProductEnvName(String productEnvName) {this.productEnvName = productEnvName;}public boolean isClusterTest() {return clusterTest;}public void setClusterTest(boolean clusterTest) {this.clusterTest = clusterTest;}public String getConfigStorePath() {return configStorePath;}public void setConfigStorePath(final String configStorePath) {this.configStorePath = configStorePath;}
}
6、org.apache.rocketmq.remoting.netty.NettyServerConfig.java
源码:
/*
D:\RocketMQ\rocketmq-master\remoting\src\main\java\org\apache\rocketmq\remoting\netty\NettyServerConfig.java* Licensed to the Apache Software Foundation (ASF) under one or more* contributor license agreements. See the NOTICE file distributed with* this work for additional information regarding copyright ownership.* The ASF licenses this file to You under the Apache License, Version 2.0* (the "License"); you may not use this file except in compliance with* the License. You may obtain a copy of the License at** http://www.apache.org/licenses/LICENSE-2.0** Unless required by applicable law or agreed to in writing, software* distributed under the License is distributed on an "AS IS" BASIS,* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.* See the License for the specific language governing permissions and* limitations under the License.*/
package org.apache.rocketmq.remoting.netty;public class NettyServerConfig implements Cloneable {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;/*** make make install*** ../glibc-2.10.1/configure \ --prefix=/usr \ --with-headers=/usr/include \* --host=x86_64-linux-gnu \ --build=x86_64-pc-linux-gnu \ --without-gd*/private boolean useEpollNativeSelector = false;public int getListenPort() {return listenPort;}public void setListenPort(int listenPort) {this.listenPort = listenPort;}public int getServerWorkerThreads() {return serverWorkerThreads;}public void setServerWorkerThreads(int serverWorkerThreads) {this.serverWorkerThreads = serverWorkerThreads;}public int getServerSelectorThreads() {return serverSelectorThreads;}public void setServerSelectorThreads(int serverSelectorThreads) {this.serverSelectorThreads = serverSelectorThreads;}public int getServerOnewaySemaphoreValue() {return serverOnewaySemaphoreValue;}public void setServerOnewaySemaphoreValue(int serverOnewaySemaphoreValue) {this.serverOnewaySemaphoreValue = serverOnewaySemaphoreValue;}public int getServerCallbackExecutorThreads() {return serverCallbackExecutorThreads;}public void setServerCallbackExecutorThreads(int serverCallbackExecutorThreads) {this.serverCallbackExecutorThreads = serverCallbackExecutorThreads;}public int getServerAsyncSemaphoreValue() {return serverAsyncSemaphoreValue;}public void setServerAsyncSemaphoreValue(int serverAsyncSemaphoreValue) {this.serverAsyncSemaphoreValue = serverAsyncSemaphoreValue;}public int getServerChannelMaxIdleTimeSeconds() {return serverChannelMaxIdleTimeSeconds;}public void setServerChannelMaxIdleTimeSeconds(int serverChannelMaxIdleTimeSeconds) {this.serverChannelMaxIdleTimeSeconds = serverChannelMaxIdleTimeSeconds;}public int getServerSocketSndBufSize() {return serverSocketSndBufSize;}public void setServerSocketSndBufSize(int serverSocketSndBufSize) {this.serverSocketSndBufSize = serverSocketSndBufSize;}public int getServerSocketRcvBufSize() {return serverSocketRcvBufSize;}public void setServerSocketRcvBufSize(int serverSocketRcvBufSize) {this.serverSocketRcvBufSize = serverSocketRcvBufSize;}public boolean isServerPooledByteBufAllocatorEnable() {return serverPooledByteBufAllocatorEnable;}public void setServerPooledByteBufAllocatorEnable(boolean serverPooledByteBufAllocatorEnable) {this.serverPooledByteBufAllocatorEnable = serverPooledByteBufAllocatorEnable;}public boolean isUseEpollNativeSelector() {return useEpollNativeSelector;}public void setUseEpollNativeSelector(boolean useEpollNativeSelector) {this.useEpollNativeSelector = useEpollNativeSelector;}@Overridepublic Object clone() throws CloneNotSupportedException {return (NettyServerConfig) super.clone();}
}
7、NameServer 启动流程 :步骤一:
解析配置文件,填充 NameServerConfig、NettyServerConfig 属性值,并创建 NamesrvController。
8、 代码: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);
9、 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;
10、 rocketmqHome: rocketmq 主目录
kvConfig: NameServer 存储KV配置属性的持久化路径
configStorePath: nameServer 默认配置文件路径
orderMessageEnable: 是否支持顺序消息
11、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、NameServer 启动流程 :步骤二
根据启动属性创建 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、NameServer 启动流程 :步骤三
在 JVM 进程关闭之前,先将线程池关闭,及时释放资源。
2、代码:NamesrvStartup#start
//注册JVM钩子函数代码
Runtime.getRuntime().addShutdownHook(new ShutdownHookThread(log, new Callable<Void>() {@Overridepublic Void call() throws Exception {//释放资源controller.shutdown();return null;}
}));
# 消息中间件 RocketMQ 高级功能和源码分析(三)
相关文章:

# 消息中间件 RocketMQ 高级功能和源码分析(四)
消息中间件 RocketMQ 高级功能和源码分析(四) 一、 消息中间件 RocketMQ 源码分析:回顾 NameServer 架构设计。 1、RocketMQ 架构设计 消息中间件的设计思路一般是基于主题订阅发布的机制,消息生产者(Producer&…...

如何通过数据库与AI实现以图搜图?OceanBase向量功能详解
OceanBase支持向量数据库的基础能力 当前,数据库存储系统与人工智能技术的结合,可以体现在两个主要的应用方向上。 一、近似搜索。它利用大语言模型(LLM,简称大模型)的嵌入(embedding)技术&am…...
Kafka内外网分流配置listeners和advertised.listeners
问题背景: Kafka部署在内网,内网Java服务会使用Kafka收发消息,另外,Java服务会与其他第三方系统使用kafka实现数据同步,也就是外网也会发送消息到kafka,外网IP做了端口映射到了内网,advertised…...

Linux系统编程——网络编程
目录 一、对于Socket、TCP/UDP、端口号的认知: 1.1 什么是Socket: 1.2 TCP/UDP对比: 1.3 端口号的作用: 二、字节序 2.1 字节序相关概念: 2.2 为什么会有字节序: 2.3 主机字节序转换成网络字节序函数…...
信息安全技术基础知识-经典题目
【第1题】 1.在信息安全领域,基本的安全性原则包括机密性(Confidentiality)、完整性(Integrity)和 可用性(Availability)。机密性指保护信息在使用、传输和存储时 (1) 。信息加密是保证系统机密性的常用手段。使用哈希校验是保证数据完整性的常用方法。可用性指保证…...

nextjs(持续学习中)
return ( <p className{${lusitana.className} text-xl text-gray-800 md:text-3xl md:leading-normal}> Welcome to Acme. This is the example for the{’ } Next.js Learn Course , brought to you by Vercel. ); } 在顶级 /public 文件夹下提供静态资产 **默认 /…...
数据预处理与特征工程、过拟合与欠拟合
数据预处理与特征工程 常用的数据预处理步骤 向量化:将数据转换成pytorch张量值归一化:将特定特征的数据表示成均值为0,标准差为1的数据的过程;取较小的值:通常在0和1之间;相同值域处理缺失值特征工程&am…...
甲辰年五月十四风雨思
甲辰年五月十四风雨思 夜雨消暑气,远光归家心。 只待万窗明,朝夕千家勤。 苦乐言行得,酸甜日常品。 宫商角徵羽,仁义礼智信。...
java分别使用 iText 7 库和iText 5 库 将excel转成PDF导出,以及如何对excel转PDF合并单元格
第一种 package com.junfun.pms.report.util;import com.itextpdf.kernel.font.PdfFontFactory; import com.itextpdf.layout.Document; import com.itextpdf.layout.element.Paragraph; import com.itextpdf.layout.property.TextAlignment; import com.itextpdf.layout.prop…...

Java特性之设计模式【访问者模式】
一、访问者模式 概述 在访问者模式(Visitor Pattern)中,我们使用了一个访问者类,它改变了元素类的执行算法。通过这种方式,元素的执行算法可以随着访问者改变而改变。这种类型的设计模式属于行为型模式。根据模式&…...
【教师资格证考试综合素质——法律专项】未成年人保护法笔记以及练习题
《中华人民共和国未成年人保护法》 目录 第一章 总 则 第二章 家庭保护 第三章 学校保护 第四章 社会保护 第五章 网络保护 第六章 政府保护 第七章 司法保护 第八章 法律责任 第九章 附 则 介一.首次颁布:第一部《中华人民共和国未成年人保护法…...
6.19作业
TCP服务器 #include <stdio.h> #include <sys/types.h> #include <sys/socket.h> #include <unistd.h> #include <arpa/inet.h> #include <netinet/in.h> #include <string.h>#define PORT 8888 #define IP "192.168.124.39&q…...

java 线程之间通信-volatile 和 synchronized
你好,我是 shengjk1,多年大厂经验,努力构建 通俗易懂的、好玩的编程语言教程。 欢迎关注!你会有如下收益: 了解大厂经验拥有和大厂相匹配的技术等 希望看什么,评论或者私信告诉我! 文章目录 一…...

资源宝库网站!人人必备的神器!
面对网络中海量的内容,一个高效、便捷的网络导航工具,可以帮助我们快速查找使用网络资源。无论是职场精英还是学生党,使用导航网站都可以帮助我们提升效率。下面小编就来和大家分享一款资源宝库网站-办公人导航-实用的办公生活导航网站&#…...

Redis实战—优惠卷秒杀(锁/事务/代理对象的应用)
本博客为个人学习笔记,学习网站与详细见:黑马程序员Redis入门到实战 P50 - P54 目录 优惠卷秒杀下单功能实现 超卖问题 悲观锁与乐观锁 实现CAS法乐观锁 一人一单功能实现 代码优化 代码细节分析 优惠卷秒杀下单功能实现 Controller层…...

HTML星空特效
目录 写在前面 完整代码 代码分析 运行效果 系列文章 写在后面 写在前面 100行代码实现HTML星空特效。 完整代码 全部代码如下。 <!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd"&g…...

银行数仓项目实战(四)--了解银行业务(存款)
文章目录 项目准备存款活期定期整存整取零存整取存本取息教育储蓄定活两便通知存款 对公存款对公账户协议存款 利率 项目准备 (贴源层不必写到项目文档,因为没啥操作没啥技术,只是数据。) 可以看到,银行的贴源层并不紧…...

MySQL版本发布模型
MySQL 8.0 之后使用了新的版本控制和发布模型,分为两个主线:长期支持版(LTS)以及创新版。这两种版本都包含了缺陷修复和安全修复,都可以用于生产环境。 下图是 MySQL 的版本发布计划: 长期支持版 MySQL…...
java: 不兼容的类型: org.apache.xmlbeans.XmlObject无法转换为x2006.main.CTRow
我使用的xmlbeans版本是5.0,使用xmlbeans包做转换时,报错,正如标题显示得那样 解决办法 额外再引入下面的jar包 <dependency><groupId>org.apache.xmlbeans</groupId><artifactId>xmlbeans</artifactId><…...

内容时代:品牌如何利用社交平台精准触达用户
还记得学生时代老师教写作文的时候常说的一句话就是“开头质量决定了阅卷老师想不想花精力去读,而内容质量决定了她愿不愿意给你判高分”这个世界仿若一个巨大的圆,同样的逻辑放在任何地方好像都能适用。在品牌营销中,内容已成为品牌与消费者…...

微信小程序之bind和catch
这两个呢,都是绑定事件用的,具体使用有些小区别。 官方文档: 事件冒泡处理不同 bind:绑定的事件会向上冒泡,即触发当前组件的事件后,还会继续触发父组件的相同事件。例如,有一个子视图绑定了b…...

WordPress插件:AI多语言写作与智能配图、免费AI模型、SEO文章生成
厌倦手动写WordPress文章?AI自动生成,效率提升10倍! 支持多语言、自动配图、定时发布,让内容创作更轻松! AI内容生成 → 不想每天写文章?AI一键生成高质量内容!多语言支持 → 跨境电商必备&am…...

涂鸦T5AI手搓语音、emoji、otto机器人从入门到实战
“🤖手搓TuyaAI语音指令 😍秒变表情包大师,让萌系Otto机器人🔥玩出智能新花样!开整!” 🤖 Otto机器人 → 直接点明主体 手搓TuyaAI语音 → 强调 自主编程/自定义 语音控制(TuyaAI…...

关键领域软件测试的突围之路:如何破解安全与效率的平衡难题
在数字化浪潮席卷全球的今天,软件系统已成为国家关键领域的核心战斗力。不同于普通商业软件,这些承载着国家安全使命的软件系统面临着前所未有的质量挑战——如何在确保绝对安全的前提下,实现高效测试与快速迭代?这一命题正考验着…...

中医有效性探讨
文章目录 西医是如何发展到以生物化学为药理基础的现代医学?传统医学奠基期(远古 - 17 世纪)近代医学转型期(17 世纪 - 19 世纪末)现代医学成熟期(20世纪至今) 中医的源远流长和一脉相承远古至…...

R 语言科研绘图第 55 期 --- 网络图-聚类
在发表科研论文的过程中,科研绘图是必不可少的,一张好看的图形会是文章很大的加分项。 为了便于使用,本系列文章介绍的所有绘图都已收录到了 sciRplot 项目中,获取方式: R 语言科研绘图模板 --- sciRplothttps://mp.…...
十九、【用户管理与权限 - 篇一】后端基础:用户列表与角色模型的初步构建
【用户管理与权限 - 篇一】后端基础:用户列表与角色模型的初步构建 前言准备工作第一部分:回顾 Django 内置的 `User` 模型第二部分:设计并创建 `Role` 和 `UserProfile` 模型第三部分:创建 Serializers第四部分:创建 ViewSets第五部分:注册 API 路由第六部分:后端初步测…...

【Veristand】Veristand环境安装教程-Linux RT / Windows
首先声明,此教程是针对Simulink编译模型并导入Veristand中编写的,同时需要注意的是老用户编译可能用的是Veristand Model Framework,那个是历史版本,且NI不会再维护,新版本编译支持为VeriStand Model Generation Suppo…...

【PX4飞控】mavros gps相关话题分析,经纬度海拔获取方法,卫星数锁定状态获取方法
使用 ROS1-Noetic 和 mavros v1.20.1, 携带经纬度海拔的话题主要有三个: /mavros/global_position/raw/fix/mavros/gpsstatus/gps1/raw/mavros/global_position/global 查看 mavros 源码,来分析他们的发布过程。发现前两个话题都对应了同一…...
算法刷题-回溯
今天给大家分享的还是一道关于dfs回溯的问题,对于这类问题大家还是要多刷和总结,总体难度还是偏大。 对于回溯问题有几个关键点: 1.首先对于这类回溯可以节点可以随机选择的问题,要做mian函数中循环调用dfs(i&#x…...