rocketmq源碼解析之NamesrvController啓動①
說在前面linux
本次解析NamesrvController的啓動過程。apache
源碼解析json
進入到這個方法org.apache.rocketmq.namesrv.NamesrvStartup#main0這一行start(controller);進入這個方法org.apache.rocketmq.namesrv.NamesrvStartup#start服務器
public static NamesrvController start(final NamesrvController controller) throws Exception {
if (null == controller) {
throw new IllegalArgumentException("NamesrvController is null");
}
// 初始化namesrv控制器 =》
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;
}
}));
// namesrv控制器啓動
controller.start();
return controller;
}
進入這個方法org.apache.rocketmq.namesrv.NamesrvController#initialize NamesrcvController初始化微信
public boolean initialize() {
// 加載配置 =》
this.kvConfigManager.load();
// 初始化netty server =》
this.remotingServer = new NettyRemotingServer(this.nettyServerConfig, this.brokerHousekeepingService);
// 初始化netty執行器 8個線程
this.remotingExecutor =
Executors.newFixedThreadPool(nettyServerConfig.getServerWorkerThreads(), new ThreadFactoryImpl("RemotingExecutorThread_"));
// 註冊處理器 =》
this.registerProcessor();
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
// 掃描非活動的broker
NamesrvController.this.routeInfoManager.scanNotActiveBroker();
}
}, 5, 10, TimeUnit.SECONDS);
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;
}
進入到這個方法，從持久化文件中加載配置到內存org.apache.rocketmq.namesrv.kvconfig.KVConfigManager#load併發
public void load() {
String content = null;
try {
// 將文件讀取成字符串 user.home/namesrv/kvConfig.json
content = MixAll.file2String(this.namesrvController.getNamesrvConfig().getKvConfigPath());
} catch (IOException e) {
log.warn("Load KV config table exception", e);
}
if (content != null) {
KVConfigSerializeWrapper kvConfigSerializeWrapper =
KVConfigSerializeWrapper.fromJson(content, KVConfigSerializeWrapper.class);
if (null != kvConfigSerializeWrapper) {
this.configTable.putAll(kvConfigSerializeWrapper.getConfigTable());
log.info("load KV config table OK");
}
}
}
進入這個方法org.apache.rocketmq.remoting.netty.NettyRemotingServer#NettyRemotingServer(org.apache.rocketmq.remoting.netty.NettyServerConfig, org.apache.rocketmq.remoting.ChannelEventListener)初始化netty服務器app
public NettyRemotingServer(final NettyServerConfig nettyServerConfig,
final ChannelEventListener channelEventListener) {
// 用信號量來控制併發數
super(nettyServerConfig.getServerOnewaySemaphoreValue(), nettyServerConfig.getServerAsyncSemaphoreValue());
this.serverBootstrap = new ServerBootstrap();
this.nettyServerConfig = nettyServerConfig;
this.channelEventListener = channelEventListener;
// 默認netty server回調線程是4
int publicThreadNums = nettyServerConfig.getServerCallbackExecutorThreads();
if (publicThreadNums <= 0) {
publicThreadNums = 4;
}
// 建立核心線程數是4個的線程池
this.publicExecutor = Executors.newFixedThreadPool(publicThreadNums, new ThreadFactory() {
private AtomicInteger threadIndex = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "NettyServerPublicExecutor_" + this.threadIndex.incrementAndGet());
}
});
// 建立有一個線程的調度線程組
this.eventLoopGroupBoss = new NioEventLoopGroup(1, new ThreadFactory() {
private AtomicInteger threadIndex = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
return new Thread(r, String.format("NettyBoss_%d", this.threadIndex.incrementAndGet()));
}
});
// 是否使用netty 的epoll組件，epoll組件在linux環境下能夠使netty事件模型性能達到最優
if (useEpoll()) {
this.eventLoopGroupSelector = new EpollEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
private AtomicInteger threadIndex = new AtomicInteger(0);
private int threadTotal = nettyServerConfig.getServerSelectorThreads();
@Override
public Thread newThread(Runnable r) {
return new Thread(r, String.format("NettyServerEPOLLSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
}
});
} else {
// selctor默認是3個線程
this.eventLoopGroupSelector = new NioEventLoopGroup(nettyServerConfig.getServerSelectorThreads(), new ThreadFactory() {
private AtomicInteger threadIndex = new AtomicInteger(0);
private int threadTotal = nettyServerConfig.getServerSelectorThreads();
@Override
public Thread newThread(Runnable r) {
return new Thread(r, String.format("NettyServerNIOSelector_%d_%d", threadTotal, this.threadIndex.incrementAndGet()));
}
});
}
// 加載ssl上下文
loadSslContext();
}
往上返回進入這個方法org.apache.rocketmq.namesrv.NamesrvController#registerProcessoride
註冊處理器oop
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);
}
}
進入這個方法org.apache.rocketmq.namesrv.processor.ClusterTestRequestProcessor#ClusterTestRequestProcessor建立集羣測試請求處理器性能
public ClusterTestRequestProcessor(NamesrvController namesrvController, String productEnvName) {
super(namesrvController);
this.productEnvName = productEnvName;
adminExt = new DefaultMQAdminExt();
adminExt.setInstanceName("CLUSTER_TEST_NS_INS_" + productEnvName);
adminExt.setUnitName(productEnvName);
try {
// 啓動消息隊列管理服務啓動=》
adminExt.start();
} catch (MQClientException e) {
log.error("Failed to start processor", e);
}
}
進入到這個方法org.apache.rocketmq.tools.admin.DefaultMQAdminExtImpl#start啓動mq默認的管理服務
@Override
public void start() throws MQClientException {
switch (this.serviceState) {
case CREATE_JUST://服務只啓動，不建立
this.serviceState = ServiceState.START_FAILED;
this.defaultMQAdminExt.changeInstanceNameToPID();
// 建立mqclient對象 =》
this.mqClientInstance = MQClientManager.getInstance().getAndCreateMQClientInstance(this.defaultMQAdminExt, rpcHook);
// 註冊管理服務處理器=》
boolean registerOK = mqClientInstance.registerAdminExt(this.defaultMQAdminExt.getAdminExtGroup(), this);
if (!registerOK) {
this.serviceState = ServiceState.CREATE_JUST;
throw new MQClientException("The adminExt group[" + this.defaultMQAdminExt.getAdminExtGroup()
+ "] has created already, specifed another name please."
+ FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL), null);
}
// 啓動mqclient =》
mqClientInstance.start();
log.info("the adminExt [{}] start OK", this.defaultMQAdminExt.getAdminExtGroup());
this.serviceState = ServiceState.RUNNING;
break;
case RUNNING:
case START_FAILED:
case SHUTDOWN_ALREADY:
throw new MQClientException("The AdminExt service state not OK, maybe started once, "
+ this.serviceState
+ FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK), null);
default:
break;
}
}
未完待續。
說在最後
本次解析僅表明我的觀點，僅供參考。
加入技術微信羣
釘釘技術羣
相關文章
- 1. rocketmq源碼解析namesrvController啓動③
- 2. rocketmq源碼解析之NamesrvController啓動②建立mqclient②
- 3. rocketmq源碼解析之NamesrvController啓動②建立mqclient①
- 4. rocketmq源碼解析之NamesrvController建立
- 5. rocketmq源碼解析之NamesrvController啓動②mqclient admin請求處理刪除topic
- 6. rocketmq源碼解析之broker啓動
- 7. rocketmq源碼解析之name啓動（一）
- 8. rocketmq源碼解析broker啓動④
- 9. rocketmq源碼解析broker啓動①
- 10. rocketmq源碼解析broker啓動③
- 更多相關文章...
- • 啓動MySQL服務 - MySQL教程
- • SVN 啓動模式 - SVN 教程
- • 互聯網組織的未來：剖析GitHub員工的任性之源
- • Scala 中文亂碼解決
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. rocketmq源碼解析namesrvController啓動③
- 2. rocketmq源碼解析之NamesrvController啓動②建立mqclient②
- 3. rocketmq源碼解析之NamesrvController啓動②建立mqclient①
- 4. rocketmq源碼解析之NamesrvController建立
- 5. rocketmq源碼解析之NamesrvController啓動②mqclient admin請求處理刪除topic
- 6. rocketmq源碼解析之broker啓動
- 7. rocketmq源碼解析之name啓動（一）
- 8. rocketmq源碼解析broker啓動④
- 9. rocketmq源碼解析broker啓動①
- 10. rocketmq源碼解析broker啓動③