Nacos1.x配置中心客户端长轮询原理

灰太狼 2024-03-29 15:42 134阅读 0赞

> nacos配置中心帮助我们很好的管理各服务中的配置文件，同时动态刷新机制很好的解决了配置文件变更后需要重启的问题。所以本文从客户端起点开始窥探具体实现原理。

#### 1. 什么是长轮询 ####

要搞懂Nacos注册中心原理，最重要的是搞清楚什么是长轮询，注册中心就是用长轮询的机制去获取最新配置信息，而长轮询和长连接有什么区别呢？

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>原理</th> 
   <th>优点</th> 
   <th>缺点</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>轮询</td> 
   <td>客户端定时向服务端请求，响应后立刻返回关闭连接</td> 
   <td>服务端编写简单，单一请求响应</td> 
   <td>浪费带宽和服务器资源</td> 
  </tr> 
  <tr> 
   <td>长轮询</td> 
   <td>客户端向服务端请求，服务端会在有新数据时候响应，否则会在一定时间后返回</td> 
   <td>避免频繁无效请求</td> 
   <td>无法保证数据顺序</td> 
  </tr> 
  <tr> 
   <td>长连接</td> 
   <td>一直保持连接不断开</td> 
   <td>消息通讯及时</td> 
   <td>维护长连接增加开销</td> 
  </tr> 
 </tbody> 
</table>

##### Nacos长轮询 #####

![在这里插入图片描述][8a13c2219f72477caaec7f3aa89ab985.png]  
从上图可以得知几个关键：

1.  nacos长轮询超时时间是30s，实际是客户端每隔30s会发送一次请求，而服务端若无数据变更会在29.5s时候响应。
2.  nacos管理后台通过推送配置方式将最新变更发送到服务端。

##### 具体细节 #####

*  客户端发起长轮询  
    客户端发起一个 HTTP 请求，请求信息包含配置中心的地址，以及监听的 dataId（本文出于简化说明的考虑，认为 dataId 是定位配置的唯一键）。若配置没有发生变化，客户端与服务端之间一直处于连接状态。
 *  服务端监听数据变化  
    服务端会维护 dataId 和长轮询的映射关系，如果配置发生变化，服务端会找到对应的连接，为响应写入更新后的配置内容。如果超时内配置未发生变化，服务端找到对应的超时长轮询连接，写入 304 响应。  
    304 在 HTTP 响应码中代表“未改变”，并不代表错误。比较契合长轮询时，配置未发生变更的场景。
 *  客户端接收长轮询响应  
    首先查看响应码是 200 还是 304，以判断配置是否变更，做出相应的回调。之后再次发起下一次长轮询。
 *  服务端设置配置写入的接入点  
    主要用配置控制台和 client 发布配置，触发配置变更

#### 2.SpringCloud配置中心原理 ####

##### 1.PropertySourceBootstrapConfiguration #####

该类是配置入口类，用于解析配置文件，我们先观察该类的继承关系。  
![在这里插入图片描述][98bd9d9320ac4cab8042f96204c54638.png]  
可以发现该类是一种ApplicationContextInitializer，而initializer是在**prepareContext**中进行初始化调用的。

可以观察到在initialize的时候会用**PropertySourceLocator**进行配置获取。

![在这里插入图片描述][833f736c20c94367a0891bc897e4211e.png]

##### 2.PropertySourceLocator #####

上文得知PropertySourceLocator就是Spring Cloud获取配置的类，那我们看一下Nacos是如何注入的。

这个时候我们可以去nacos config自动配置文件里找一下。  
![在这里插入图片描述][a3b2c69e2d8645b086c801a795a57ec8.png]

果然，我们可以看到在NacosConfigBootstrapConfiguration配置了我们需要的nacos实现类NacosPropertySourceLocator。接着我们继续往其实现方法locate中看。

##### 3.NacosConfigService #####

从locate往里进入最后定位到ConfigFactory::createConfigService，顾名思义该方法用于创建配置中心。  
![在这里插入图片描述][2e1adabc617b4a2485136e86d5bbe199.png]  
所以最关键nacos配置中心核心类就是**NacosConfigService**，configfactory通过反射方式创建NacosConfigService，所以我们这个时候可以往该类的构造方法去看。

public NacosConfigService(Properties properties) throws NacosException {
        String encodeTmp = properties.getProperty("encode");
        if (StringUtils.isBlank(encodeTmp)) {
            this.encode = "UTF-8";
        } else {
            this.encode = encodeTmp.trim();
        }
        // 初始化命名空间
        this.initNamespace(properties);
        //初始化 HttpAgent，用到了装饰器模式，实际工作的类是 ServerHttpAgent
        this.agent = new MetricsHttpAgent(new ServerHttpAgent(properties));
        this.agent.start();
      	// 核心类，用于配置拉取和刷新
        this.worker = new ClientWorker(this.agent, this.configFilterChainManager, properties);
    }

##### 4.ClientWorker #####

public ClientWorker(final HttpAgent agent, ConfigFilterChainManager configFilterChainManager, Properties properties) {
        this.agent = agent;
        this.configFilterChainManager = configFilterChainManager;
        this.init(properties);
        //创建 executor 线程池，只拥有一个核心线程，每隔 10ms 就会执行一次 checkConfiglnfo() 方法，检查配置信息
        this.executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("com.alibaba.nacos.client.Worker." + agent.getName());
                t.setDaemon(true);
                return t;
            }
        });
        //创建 executorService 线程池，只完成了初始化，后续会用到，主要用于实现客户端的定时长轮询功能
        this.executorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("com.alibaba.nacos.client.Worker.longPolling." + agent.getName());
                t.setDaemon(true);
                return t;
            }
        });
        //使用 executor 启动一个每隔 10ms 执行一次的定时任务
        this.executor.scheduleWithFixedDelay(new Runnable() {
            public void run() {
                try {
                    // 检查配置是否发生变化
                    ClientWorker.this.checkConfigInfo();
                } catch (Throwable var2) {
                    ClientWorker.LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", var2);
                }
            }
        }, 1L, 10L, TimeUnit.MILLISECONDS);
    }

两个线程池，一个负责检查配置是否更新，一个长轮询进行拉取。

接着我们可以看到启动的runnable里面的run方法。

public void run() {
        List<CacheData> cacheDatas = new ArrayList();
        ArrayList inInitializingCacheList = new ArrayList();
        try {
            //遍历 CacheData，检查本地配置
            Iterator var3 = ((Map)ClientWorker.this.cacheMap.get()).values().iterator();
            while(var3.hasNext()) {
                CacheData cacheData = (CacheData)var3.next();
                if (cacheData.getTaskId() == this.taskId) {
                    cacheDatas.add(cacheData);
                    try {
                        //检查本地配置
                        ClientWorker.this.checkLocalConfig(cacheData);
                        if (cacheData.isUseLocalConfigInfo()) {
                            cacheData.checkListenerMd5();
                        }
                    } catch (Exception var13) {
                        ClientWorker.LOGGER.error("get local config info error", var13);
                    }
                }
            }
            // 核心1：通过长轮询请求检查服务端对应的配置是否发生变更
            List<String> changedGroupKeys = ClientWorker.this.checkUpdateDataIds(cacheDatas, inInitializingCacheList);
            //遍历存在变更的 groupKey，重新加载最新数据
            Iterator var16 = changedGroupKeys.iterator();
            while(var16.hasNext()) {
                String groupKey = (String)var16.next();
                String[] key = GroupKey.parseKey(groupKey);
                String dataId = key[0];
                String group = key[1];
                String tenant = null;
                if (key.length == 3) {
                    tenant = key[2];
                }
                try {
                    // 核心2：根据dataId、group 和 tenant 进行获取配置
                    String content = ClientWorker.this.getServerConfig(dataId, group, tenant, 3000L);
                    CacheData cache = (CacheData)((Map)ClientWorker.this.cacheMap.get()).get(GroupKey.getKeyTenant(dataId, group, tenant));
                    cache.setContent(content);
                    ClientWorker.LOGGER.info("[{}] [data-received] dataId={}, group={}, tenant={}, md5={}, content={}", new Object[]{ClientWorker.this.agent.getName(), dataId, group, tenant, cache.getMd5(), ContentUtils.truncateContent(content)});
                } catch (NacosException var12) {
                    String message = String.format("[%s] [get-update] get changed config exception. dataId=%s, group=%s, tenant=%s", ClientWorker.this.agent.getName(), dataId, group, tenant);
                    ClientWorker.LOGGER.error(message, var12);
                }
            }
            //触发事件通知
            var16 = cacheDatas.iterator();
            while(true) {
                CacheData cacheDatax;
                do {
                    if (!var16.hasNext()) {
                        inInitializingCacheList.clear();
                        //继续定时执行当前线程
                        ClientWorker.this.executorService.execute(this);
                        return;
                    }
                    cacheDatax = (CacheData)var16.next();
                } while(cacheDatax.isInitializing() && !inInitializingCacheList.contains(GroupKey.getKeyTenant(cacheDatax.dataId, cacheDatax.group, cacheDatax.tenant)));
                cacheDatax.checkListenerMd5();
                cacheDatax.setInitializing(false);
            }
        } catch (Throwable var14) {
            ClientWorker.LOGGER.error("longPolling error : ", var14);
            ClientWorker.this.executorService.schedule(this, (long)ClientWorker.this.taskPenaltyTime, TimeUnit.MILLISECONDS);
        }
    }

上面的线程主要做了两件核心事

1.  通过服务端长轮询检查更新了的配置。
2.  对更新了的服务配置进行配置获取。

##### 1.ClientWorker.checkUpdateDataIds() #####

我们点进 ClientWorker.checkUpdateDataIds() 方法，发现其最终调用的是 ClientWorker.checkUpdateConfigStr() 方法，其实现逻辑与源码如下：

List<String> checkUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) throws IOException {
        List<String> params = Arrays.asList("Listening-Configs", probeUpdateString);
        List<String> headers = new ArrayList(2);
        headers.add("Long-Pulling-Timeout");
        headers.add("" + this.timeout);
        if (isInitializingCacheList) {
            headers.add("Long-Pulling-Timeout-No-Hangup");
            headers.add("true");
        }
        if (StringUtils.isBlank(probeUpdateString)) {
            return Collections.emptyList();
        } else {
            try {
                //调用 /v1/cs/configs/listener 接口实现长轮询请求，返回的 HttpResult 里包含存在数据变更的 Data ID、Group、Tenant
                HttpResult result = this.agent.httpPost("/v1/cs/configs/listener", headers, params, this.agent.getEncode(), this.timeout);
                if (200 == result.code) {
                    this.setHealthServer(true);
                    //
                    return this.parseUpdateDataIdResponse(result.content);
                }
                this.setHealthServer(false);
                LOGGER.error("[{}] [check-update] get changed dataId error, code: {}", this.agent.getName(), result.code);
            } catch (IOException var6) {
                this.setHealthServer(false);
                LOGGER.error("[" + this.agent.getName() + "] [check-update] get changed dataId exception", var6);
                throw var6;
            }
            return Collections.emptyList();
        }
    }

*  通过 `MetricsHttpAgent.httpPost()` 方法（上面 1.2.1 有提到）调用 `/v1/cs/configs/listener` 接口实现长轮询请求；
 *  长轮询请求在实现层面只是设置了一个比较长的超时时间，默认是 30s；
 *  如果服务端的数据发生了变更，客户端会收到一个 HttpResult ，服务端返回的是存在数据变更的 Data ID、Group、Tenant；
 *  获得这些信息之后，在 `LongPollingRunnable.run()` 方法中调用 getServerConfig() 去 Nacos 服务器上读取具体的配置内容；

##### 2.ClientWorker.getServerConfig() #####

获取对应配置并保存到本地。

public String getServerConfig(String dataId, String group, String tenant, long readTimeout) throws NacosException {
        if (StringUtils.isBlank(group)) {
            group = "DEFAULT_GROUP";
        }
        HttpResult result = null;
        try {
            List<String> params = null;
            if (StringUtils.isBlank(tenant)) {
                params = Arrays.asList("dataId", dataId, "group", group);
            } else {
                params = Arrays.asList("dataId", dataId, "group", group, "tenant", tenant);
            }
            //获取变更配置的接口调用
            result = this.agent.httpGet("/v1/cs/configs", (List)null, params, this.agent.getEncode(), readTimeout);
        } catch (IOException var9) {
            String message = String.format("[%s] [sub-server] get server config exception, dataId=%s, group=%s, tenant=%s", this.agent.getName(), dataId, group, tenant);
            LOGGER.error(message, var9);
            throw new NacosException(500, var9);
        }
        switch(result.code) {
        //获取变更的配置成功，添加进缓存里
        case 200:
            LocalConfigInfoProcessor.saveSnapshot(this.agent.getName(), dataId, group, tenant, result.content);
            //result.content 就是我们变更后的配置信息
            return result.content;
        case 403:
            LOGGER.error("[{}] [sub-server-error] no right, dataId={}, group={}, tenant={}", new Object[]{this.agent.getName(), dataId, group, tenant});
            throw new NacosException(result.code, result.content);
        case 404:
            LocalConfigInfoProcessor.saveSnapshot(this.agent.getName(), dataId, group, tenant, (String)null);
            return null;
        case 409:
            LOGGER.error("[{}] [sub-server-error] get server config being modified concurrently, dataId={}, group={}, tenant={}", new Object[]{this.agent.getName(), dataId, group, tenant});
            throw new NacosException(409, "data being modified, dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
        default:
            LOGGER.error("[{}] [sub-server-error]  dataId={}, group={}, tenant={}, code={}", new Object[]{this.agent.getName(), dataId, group, tenant, result.code});
            throw new NacosException(result.code, "http error, code=" + result.code + ",dataId=" + dataId + ",group=" + group + ",tenant=" + tenant);
        }
    }

#### 总结： ####

Nacos1.x选择长轮询的方式实现配置中心，很巧妙的方式避免短连接频繁请求服务端的弊端，同时还根据nacos的多层配置设计，很好的进行配置划分。核心通过ClientWorker类对客户端的配置变更，配置获取进行处理。

[8a13c2219f72477caaec7f3aa89ab985.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/6e389bec58fe4660b4b362c057edb586.png
[98bd9d9320ac4cab8042f96204c54638.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/711dc27f663841ae872ddb32f5ed0678.png
[833f736c20c94367a0891bc897e4211e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/b1ab4193973c4e1cbc287f02ffa8c54a.png
[a3b2c69e2d8645b086c801a795a57ec8.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/17fc204a6b7c4949bcbb5d25c7f2bc5a.png
[2e1adabc617b4a2485136e86d5bbe199.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/28/c4e91516a2014c61a5efe0fb4c442aa2.png