SpringBoot定時監(jiān)聽RocketMQ的NameServer問題及解決方案
問題分析
- 自己在測試環(huán)境部署了RocketMQ,發(fā)現(xiàn)namesrv很容易掛掉,于是就想著監(jiān)控,掛了就發(fā)郵件通知。
- 查看了rocketmq-dashboard項目,發(fā)現(xiàn)只能監(jiān)控Broker,遂放棄這一路徑。
- 于是就從報錯的日志入手,發(fā)現(xiàn)最終可以根據(jù)RocketMQTemplate獲得可活動的NameServer。
報錯日志
報錯日志如下:
12月 25 13:59:22 192.168.240.65 java[59571]: 2023-12-25 13:59:22.598 INFO 59571 --- [tWorkerThread_2] RocketmqRemoting : NETTY CLIENT PIPELINE: CLOSE 192.168.240.86:9876
12月 25 13:59:22 192.168.240.65 java[59571]: 2023-12-25 13:59:22.598 INFO 59571 --- [tWorkerThread_2] RocketmqRemoting : closeChannel: the channel[192.168.240.86:9876] was removed from channel table
12月 25 13:59:22 192.168.240.65 java[59571]: 2023-12-25 13:59:22.598 INFO 59571 --- [tWorkerThread_2] RocketmqRemoting : NETTY CLIENT PIPELINE: CLOSE 192.168.240.86:9876
12月 25 13:59:22 192.168.240.65 java[59571]: 2023-12-25 13:59:22.598 INFO 59571 --- [tWorkerThread_2] RocketmqRemoting : eventCloseChannel: the channel[null] has been removed from the channel table before
12月 25 13:59:22 192.168.240.65 java[59571]: 2023-12-25 13:59:22.598 INFO 59571 --- [lientSelector_1] RocketmqRemoting : closeChannel: close the connection to remote address[192.168.240.86:9876] result: true
12月 25 13:59:25 192.168.240.65 java[59571]: 2023-12-25 13:59:25.597 INFO 59571 --- [ntScan_thread_1] RocketmqRemoting : createChannel: begin to connect remote host[192.168.240.86:9876] asynchronously
12月 25 13:59:25 192.168.240.65 java[59571]: 2023-12-25 13:59:25.597 INFO 59571 --- [tWorkerThread_3] RocketmqRemoting : NETTY CLIENT PIPELINE: CONNECT UNKNOWN => 192.168.240.86:9876
12月 25 13:59:25 192.168.240.65 java[59571]: 2023-12-25 13:59:25.598 WARN 59571 --- [ntScan_thread_1] RocketmqRemoting : createChannel: connect remote host[192.168.240.86:9876] failed, AbstractBootstrap$PendingRegistrationPromise@f2a3fc5(failure: io.netty.channel.AbstractChannel$AnnotatedConnectException: 拒絕連接: /192.168.240.86:9876)
根據(jù)日志可以發(fā)現(xiàn)是NettyRemotingClient類在做監(jiān)控,持續(xù)調(diào)用,具體核心方法:
org.apache.rocketmq.remoting.netty.NettyRemotingClient#createChannel
createChannel的源碼:
private Channel createChannel(String addr) throws InterruptedException { NettyRemotingClient.ChannelWrapper cw = (NettyRemotingClient.ChannelWrapper)this.channelTables.get(addr); if (cw != null && cw.isOK()) { return cw.getChannel(); } else { if (this.lockChannelTables.tryLock(3000L, TimeUnit.MILLISECONDS)) { try { cw = (NettyRemotingClient.ChannelWrapper)this.channelTables.get(addr); boolean createNewConnection; if (cw != null) { if (cw.isOK()) { Channel var4 = cw.getChannel(); return var4; } if (!cw.getChannelFuture().isDone()) { createNewConnection = false; } else { this.channelTables.remove(addr); createNewConnection = true; } } else { createNewConnection = true; } if (createNewConnection) { ChannelFuture channelFuture = this.bootstrap.connect(RemotingHelper.string2SocketAddress(addr)); LOGGER.info("createChannel: begin to connect remote host[{}] asynchronously", addr); cw = new NettyRemotingClient.ChannelWrapper(channelFuture); this.channelTables.put(addr, cw); } } catch (Exception var8) { LOGGER.error("createChannel: create channel exception", var8); } finally { this.lockChannelTables.unlock(); } } else { LOGGER.warn("createChannel: try to lock channel table, but timeout, {}ms", 3000L); } if (cw != null) { ChannelFuture channelFuture = cw.getChannelFuture(); if (channelFuture.awaitUninterruptibly((long)this.nettyClientConfig.getConnectTimeoutMillis())) { if (cw.isOK()) { LOGGER.info("createChannel: connect remote host[{}] success, {}", addr, channelFuture.toString()); return cw.getChannel(); } LOGGER.warn("createChannel: connect remote host[" + addr + "] failed, " + channelFuture.toString()); } else { LOGGER.warn("createChannel: connect remote host[{}] timeout {}ms, {}", new Object[]{addr, this.nettyClientConfig.getConnectTimeoutMillis(), channelFuture.toString()}); } } return null; } }
從源碼中可以看到報錯的日志數(shù)據(jù)
追溯
以NettyRemotingClient類為起點,使用Debug分析,最終可以看到完整的調(diào)用鏈路:
監(jiān)控開發(fā)
那么監(jiān)控開發(fā)就很容易了,注冊RocketMQTemplate,使用定時任務(wù)監(jiān)聽即可,示例代碼如下:
@Slf4j @Component public class MQMonitorTask { @Resource private RocketMQTemplate rocketMQTemplate; @Scheduled(cron = "0/10 * * * * ?") public void scanNameServerBroker() { org.apache.rocketmq.remoting.RemotingClient remotingClient = rocketMQTemplate.getProducer() .getDefaultMQProducerImpl().getMqClientFactory().getMQClientAPIImpl().getRemotingClient(); // 注冊的 NameServer List<String> nameServerAddressList = remotingClient.getNameServerAddressList(); // 當前活躍的 NameServer List<String> availableNameSrvList = remotingClient.getAvailableNameSrvList(); log.info("nameServerAddressList:{}", JSONUtil.toJsonStr(nameServerAddressList)); log.info("availableNameSrvList:{}", JSONUtil.toJsonStr(availableNameSrvList)); // 只要 nameServerAddressList 和 availableNameSrvList 大小不一致,即可做郵件通知,具體閾值自己設(shè)置?。。? // TODO:郵件通知 } }
另外要在SprongBoot啟動類加上注解@EnableScheduling來開啟定時任務(wù)。
到此這篇關(guān)于SpringBoot定時監(jiān)聽RocketMQ的NameServer的文章就介紹到這了,更多相關(guān)SpringBoot定時監(jiān)聽NameServer內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Swagger2不被SpringSecurity框架攔截的配置及說明
這篇文章主要介紹了Swagger2不被SpringSecurity框架攔截的配置及說明,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2023-03-03Springboot?application.yml配置文件拆分方式
這篇文章主要介紹了Springboot?application.yml配置文件拆分方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-05-05Druid連接池未關(guān)閉導致內(nèi)存泄漏問題
這篇文章主要介紹了Druid連接池未關(guān)閉導致內(nèi)存泄漏問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-12-12IDEA POM文件配置profile實現(xiàn)不同環(huán)境切換的方法步驟
這篇文章主要介紹了IDEA POM文件配置profile實現(xiàn)不同環(huán)境切換的方法步驟2024-03-03