探秘RocketMQ 5.0 client端Nameserver地址更新的源码实现方式

爱被打了一巴掌 2023-09-24 23:04 3阅读 0赞

### 源码版本 ###

*  rocketmq:release-5.0.0

### 背景 ###

上一章我们通过源码的方式详细的介绍了client如何获取Nameserver地址，今天我们就来继续研究研究client是如何更新Nameserver地址

### Client如何更新本地namesrvAddr ###

client更新Nameserver的定时器主要是在这里启动的

this.startScheduledTask();
    复制代码

![format_png][]

我们进去这个方法看一看

![format_png 1][]

如果clientConfig获取的namesrvAddr地址为空就会启动一个定时任务

其中定时任务中方法

MQClientInstance.this.mQClientAPIImpl.fetchNameServerAddr();
    复制代码

就是通过我们设置的wsAddr(也就是获取namesrvAddr的url地址)地址去获取namesrvAddr地址

线程池scheduledExecutorService的初始化

privatefinalScheduledExecutorServicescheduledExecutorService= Executors.newSingleThreadScheduledExecutor(r -> newThread(r, "MQClientFactoryScheduledThread"));
    复制代码

可以看到是线程池延迟1s后执行，每隔2分钟去http服务器去拉取namesrvAddr的地址

实际更新本地的namesrvAddr地址是通过updateNameServerAddressList方法

![format_png 2][]

最终实际去更新的还是NettyRemotingClient中的namesrvAddrList值

![format_png 3][]

@OverridepublicvoidupdateNameServerAddressList(List<String> addrs) {
            List<String> old = this.namesrvAddrList.get();
            booleanupdate=false;
    
            if (!addrs.isEmpty()) {
                if (null == old) {
                    update = true;
                } elseif (addrs.size() != old.size()) {
                    update = true;
                } else {
                    for (inti=0; i < addrs.size() && !update; i++) {
                        if (!old.contains(addrs.get(i))) {
                            update = true;
                        }
                    }
                }
    
                if (update) {
                    Collections.shuffle(addrs);
                    LOGGER.info("name server address updated. NEW : {} , OLD: {}", addrs, old);
                    this.namesrvAddrList.set(addrs);
    
                    // should close the channel if choosed addr is not exist.if (this.namesrvAddrChoosed.get() != null && !addrs.contains(this.namesrvAddrChoosed.get())) {
                        StringnamesrvAddr=this.namesrvAddrChoosed.get();
                        for (String addr : this.channelTables.keySet()) {
                            if (addr.contains(namesrvAddr)) {
                                ChannelWrapperchannelWrapper=this.channelTables.get(addr);
                                if (channelWrapper != null) {
                                    closeChannel(channelWrapper.getChannel());
                                }
                            }
                        }
                    }
                }
            }
        }
    复制代码

这里除了更新namesrvAddrList的地址外，还通过Collections.shuffle(addrs)方法对addrs的地址进行了随机打散，就是为了后面再随机选用一个namesrvAddr地址，同时还做了一个事情就是如果新获取到的namesrvAddr和原有的对比，如果旧的namesrvAddr地址有被新namesrvAddr地址剔除的就会将客户端与旧NameServer断开tcp连接

closeChannel(channelWrapper.getChannel());
    复制代码

值得注意的这里都是客户端的namesrvAddr更新，还没有涉及到namesrvAddr的重连，所以我们还需要看看namesrvAddr重连的定时器

跟随源码可以定位到重连的定时器主要在NettyRemotingClient这个类中启动的

*  NettyRemotingClient.this.scanAvailableNameSrv()

![format_png 4][]

其中timer的初始化也是new了一个简单的Timer

privatefinalTimertimer=newTimer("ClientHouseKeepingService", true);
    复制代码

可以看到是启动就直接执行没有延时时间，循环时间默认3s执行一次，可通过参数com.rocketmq.remoting.client.connect.timeout设置

![format_png 5][]

这里我们进入到scanAvailableNameSrv方法看看

privatevoidscanAvailableNameSrv() {
            List<String> nameServerList = this.namesrvAddrList.get();
            if (nameServerList == null) {
                LOGGER.debug("scanAvailableNameSrv Addresses of name server is empty!");
                return;
            }
    
            for (final String namesrvAddr : nameServerList) {
                scanExecutor.execute(newRunnable() {
                    @Overridepublicvoidrun() {
                        try {
                            Channelchannel= NettyRemotingClient.this.getAndCreateChannel(namesrvAddr);
                            if (channel != null) {
                                NettyRemotingClient.this.availableNamesrvAddrMap.putIfAbsent(namesrvAddr, true);
                            } else {
                                NettyRemotingClient.this.availableNamesrvAddrMap.remove(namesrvAddr);
                            }
                        } catch (Exception e) {
                            LOGGER.error("scanAvailableNameSrv get channel of {} failed, ", namesrvAddr, e);
                        }
                    }
                });
            }
    
        }
    复制代码

可以看到这里面又通过一个线程池scanExecutor去异步扫描多个namesrvAddr

我们这里看看scanExecutor线程池的初始化配置

this.scanExecutor = newThreadPoolExecutor(4, 10, 60, TimeUnit.SECONDS,
                newArrayBlockingQueue<Runnable>(32), newThreadFactory() {
                    privatefinalAtomicIntegerthreadIndex=newAtomicInteger(0);
    
                    @Overridepublic Thread newThread(Runnable r) {
                        returnnewThread(r, "NettyClientScan_thread_" + this.threadIndex.incrementAndGet());
                    }
                }
            );
    复制代码

我们进入NettyRemotingClient.this.getAndCreateChannel方法看看

![format_png 6][]

注意这里第一行有一个判断就是

null == addr
    复制代码

这个判断是为了MQClientInstance.this.updateTopicRouteInfoFromNameServer();这里准备的

还记得我们的startScheduledTask()方法吗这里是开了一个定时器，通过方法MQClientInstance.this.updateTopicRouteInfoFromNameServer();定时去更新topic的信息

![format_png 7][]

默认时间可以看到是30s

![format_png 8][]

我们可以看看TopicRouteData的相关数据

![format_png 9][]

值得一提的我们发送消息最终使用的topic、queue相关信息是使用的TopicPublishInfo属性，TopicRouteData与TopicPublishInfo相关的转化是通过public static TopicPublishInfo topicRouteData2TopicPublishInfo(final String topic, final TopicRouteData route)方法

![format_png 10][]

我们可以看看转换后的TopicPublishInfo是啥样

![format_png 11][]

然后我们来看看getAndCreateNameserverChannel()方法

private Channel getAndCreateNameserverChannel()throws InterruptedException {
            Stringaddr=this.namesrvAddrChoosed.get();
            if (addr != null) {
                ChannelWrappercw=this.channelTables.get(addr);
                if (cw != null && cw.isOK()) {
                    return cw.getChannel();
                }
            }
    
            final List<String> addrList = this.namesrvAddrList.get();
            if (this.namesrvChannelLock.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                try {
                    addr = this.namesrvAddrChoosed.get();
                    if (addr != null) {
                        ChannelWrappercw=this.channelTables.get(addr);
                        if (cw != null && cw.isOK()) {
                            return cw.getChannel();
                        }
                    }
    
                    if (addrList != null && !addrList.isEmpty()) {
                        for (inti=0; i < addrList.size(); i++) {
                            intindex=this.namesrvIndex.incrementAndGet();
                            index = Math.abs(index);
                            index = index % addrList.size();
                            StringnewAddr= addrList.get(index);
    
                            this.namesrvAddrChoosed.set(newAddr);
                            LOGGER.info("new name server is chosen. OLD: {} , NEW: {}. namesrvIndex = {}", addr, newAddr, namesrvIndex);
                            ChannelchannelNew=this.createChannel(newAddr);
                            if (channelNew != null) {
                                return channelNew;
                            }
                        }
                        thrownewRemotingConnectException(addrList.toString());
                    }
                } catch (Exception e) {
                    LOGGER.error("getAndCreateNameserverChannel: create name server channel exception", e);
                } finally {
                    this.namesrvChannelLock.unlock();
                }
            } else {
                LOGGER.warn("getAndCreateNameserverChannel: try to lock name server, but timeout, {}ms", LOCK_TIMEOUT_MILLIS);
            }
    
            returnnull;
        }
    复制代码

有一些值得注意的是

1.  通过namesrvAddrChoosed获取nameserver地址

1.  连接的channal存储在channelTables中复用

1.  如果namesrvAddrChoosed获取到的nameserver地址不存在或者不是存活的则随机再从namesrvAddrList获取一个地址具体核心代码如下

intindex=this.namesrvIndex.incrementAndGet();
                            index = Math.abs(index);
                            index = index % addrList.size();
                            StringnewAddr= addrList.get(index);
    
                            this.namesrvAddrChoosed.set(newAddr);
    复制代码

> 这里就是每次发送消息更新topic的流程

然后我们来看看scanAvailableNameSrv流程的this.createChannel(addr)方法

private Channel createChannel(final String addr)throws InterruptedException {
    	//获取原来的ChannelWrapperChannelWrappercw=this.channelTables.get(addr);
    	// 如果存在并且存活则直接返回if (cw != null && cw.isOK()) {
                return cw.getChannel();
            }
    
            if (this.lockChannelTables.tryLock(LOCK_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS)) {
                try {
    		// 是否创建新的连接boolean createNewConnection;
    		//双重检查 类似双重检查的单例模式
                    cw = this.channelTables.get(addr);
                    if (cw != null) {
    
                        if (cw.isOK()) {
                            return cw.getChannel();
                        } elseif (!cw.getChannelFuture().isDone()) {
                            createNewConnection = false;
                        } else {
                            this.channelTables.remove(addr);
                            createNewConnection = true;
                        }
                    } else {
                        createNewConnection = true;
                    }
    		            // 创建新的连接if (createNewConnection) {
    		                // 连接到 NameserverChannelFuturechannelFuture=this.bootstrap.connect(RemotingHelper.string2SocketAddress(addr));
                        LOGGER.info("createChannel: begin to connect remote host[{}] asynchronously", addr);
                        cw = newChannelWrapper(channelFuture);
                        this.channelTables.put(addr, cw);
                    }
                } catch (Exception e) {
                    LOGGER.error("createChannel: create channel exception", e);
                } finally {
                    this.lockChannelTables.unlock();
                }
            } else {
                LOGGER.warn("createChannel: try to lock channel table, but timeout, {}ms", LOCK_TIMEOUT_MILLIS);
            }
    
            if (cw != null) {
                ChannelFuturechannelFuture= cw.getChannelFuture();
                if (channelFuture.awaitUninterruptibly(this.nettyClientConfig.getConnectTimeoutMillis())) {
                    if (cw.isOK()) {
                        LOGGER.info("createChannel: connect remote host[{}] success, {}", addr, channelFuture.toString());
                        return cw.getChannel();
                    } else {
                        LOGGER.warn("createChannel: connect remote host[" + addr + "] failed, " + channelFuture.toString());
                    }
                } else {
                    LOGGER.warn("createChannel: connect remote host[{}] timeout {}ms, {}", addr, this.nettyClientConfig.getConnectTimeoutMillis(),
                        channelFuture.toString());
                }
            }
    
            returnnull;
        }
    复制代码

### 小结 ###

总的来说客户端获取namesrvAddr的方式是三秒更新一次，所以客户端3s就能感受到namesrvAddr是否下线。

连接相关的信息存储在channelTables中，即

![format_png 12][]

### 总结 ###

1.  客户端每隔30s就会去扫描判断Nameserver是否存活

1.  客户端每隔2分钟才会去http服务器拉取最新的Nameserver地址

1.  客户端topic相关的缓存信息每隔30s会去Nameserver拉取最新的topic、broker相关信息

作者：weihubeats

链接：[https://juejin.cn/post/7208793045479456828][https_juejin.cn_post_7208793045479456828]

[format_png]: https://img-blog.csdnimg.cn/img_convert/465e399dc1a64c1feed97901a5c779a2.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/e7e11718c8ad472a520eec69ea4dfedc.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/190494393f77c3324ba9f0c5fe37901b.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/161be03db403d0df3a0fa9f9a48c851f.webp?x-oss-process=image/format,png
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/0373bbbce89413be7e067cfa250219e0.webp?x-oss-process=image/format,png
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/348b4bc2750ae32dc743223ba2b38756.webp?x-oss-process=image/format,png
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/8ae2051d0c79a38a355e83126d138db8.webp?x-oss-process=image/format,png
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/61b6d76acadda44500b25061e4e59b4e.webp?x-oss-process=image/format,png
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/a2bfc1a4782d1112f59bd2d862c45ad7.webp?x-oss-process=image/format,png
[format_png 9]: https://img-blog.csdnimg.cn/img_convert/81767c0b77730b29c496ab56033b313a.webp?x-oss-process=image/format,png
[format_png 10]: https://img-blog.csdnimg.cn/img_convert/7cc2b6f29553de2a011539b4399b1eda.webp?x-oss-process=image/format,png
[format_png 11]: https://img-blog.csdnimg.cn/img_convert/2319df2e2e73830ca3ddda569384659a.webp?x-oss-process=image/format,png
[format_png 12]: https://img-blog.csdnimg.cn/img_convert/6591302d5b330c14a476eab3109a9d2f.webp?x-oss-process=image/format,png
[https_juejin.cn_post_7208793045479456828]: https://juejin.cn/post/7208793045479456828