SpringBoot定时监听RocketMQ的NameServer问题及解决方案

org.apache.rocketmq.remoting.netty.NettyRemotingClient#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;
        }
    }

@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 大小不一致，即可做邮件通知，具体阈值自己设置！！！
        // TODO:邮件通知
    }
}