Nacos 配置服务客户端源码分析（一）

客官°小女子只卖身不卖艺 2023-09-25 18:38 64阅读 0赞

Nacos除了服务注册中心，还集成了配置中心。本篇就从Nacos的配置中心的客户端源码入手。分析Nacos客户端是如何获取到Nacos服务的配置文件的。

分析源码的第一步是找到入口，官方再次给了我们一个注册中心的样例代码，在源码的nacos-example工程中。先看下这部分的代码。

publicstaticvoidmain(String[] args)throws NacosException, InterruptedException {
        StringserverAddr="localhost";
        StringdataId="test";
        Stringgroup="DEFAULT_GROUP";
        Propertiesproperties=newProperties();
        properties.put("serverAddr", serverAddr);
        // 通过工厂模式创建了一个ConfigService服务
        ConfigServiceconfigService= NacosFactory.createConfigService(properties);
        // 获取配置信息
        Stringcontent= configService.getConfig(dataId, group, 5000);
        System.out.println(content);
        // 添加监听器，监听对应dataId，group的配置信息
        configService.addListener(dataId, group, newListener() {
            @Override
            publicvoidreceiveConfigInfo(String configInfo) {
                System.out.println("receive:" + configInfo);
            }
            @Override
            public Executor getExecutor() {
                returnnull;
            }
        });
        // 发布配置信息
        booleanisPublishOk= configService.publishConfig(dataId, group, "content");
        System.out.println(isPublishOk);
        Thread.sleep(3000);
        // 获取配置
        content = configService.getConfig(dataId, group, 5000);
        System.out.println(content);
        // 移除配置
        booleanisRemoveOk= configService.removeConfig(dataId, group);
        System.out.println(isRemoveOk);
        Thread.sleep(3000);
        // 获取配置
        content = configService.getConfig(dataId, group, 5000);
        System.out.println(content);
        Thread.sleep(300000);
    }
    复制代码

## NacosConfigService创建 ##

创建ConfigService实现比较简单，就是通过反射创建了NacosConfigService。只是要注意下创建NacosConfigService时会调用它的构造方法，做一些初始化的操作。

publicNacosConfigService(Properties properties)throws NacosException {
        finalNacosClientPropertiesclientProperties= NacosClientProperties.PROTOTYPE.derive(properties);
        ValidatorUtils.checkInitParam(clientProperties);
        // 初始化Namespace
        initNamespace(clientProperties);
        // 创建了一个配置过滤器链，可以采用SPI扩展机制加载对应的过滤器实现类
        this.configFilterChainManager = newConfigFilterChainManager(clientProperties.asProperties());
        ServerListManagerserverListManager=newServerListManager(clientProperties);
        // 创建了一个服务管理器，内含一个定时轮询线程池，每隔30s拉取一次服务
        serverListManager.start();
        // 创建了一个客户端工作者，包含了一个代理对象
        this.worker = newClientWorker(this.configFilterChainManager, serverListManager, clientProperties);
        // will be deleted in 2.0 later versions
        agent = newServerHttpAgent(serverListManager);
    }
    复制代码

在下一个篇章将重点介绍一下ClientWorker,这个是个核心，所有和服务端的交互都通过它。因为本篇主要讲解客户端注册的流程框架，先不深入细节。这里只需要知道这里进行了创建。

## 获取配置信息 ##

创建ConfigService后就可以通过接口获取配置信息了。

Stringcontent= configService.getConfig(dataId, group, 5000);
    
    public String getConfig(String dataId, String group, long timeoutMs)throws NacosException {
        return getConfigInner(namespace, dataId, group, timeoutMs);
    }
    
    private String getConfigInner(String tenant, String dataId, String group, long timeoutMs)throws NacosException {
        group = blank2defaultGroup(group);
        // 检查参数
        ParamUtils.checkKeyParam(dataId, group);
        ConfigResponsecr=newConfigResponse();
        // 设置配置信息
        cr.setDataId(dataId);
        cr.setTenant(tenant);
        cr.setGroup(group);
    
        // We first try to use local failover content if exists.
        // A config content for failover is not created by client program automatically,
        // but is maintained by user.
        // This is designed for certain scenario like client emergency reboot,
        // changing config needed in the same time, while nacos server is down.
        // 这里有个失败转移的配置。如果能读到失败转移的配置信息，则直接返回了。原因的话英文注释写的很清楚了
        // 优先使用失败转移，设计的目的是当server挂后，又需要修改配置，就可以读本地目录
        Stringcontent= LocalConfigInfoProcessor.getFailover(worker.getAgentName(), dataId, group, tenant);
        if (content != null) {
            LOGGER.warn("[{}] [get-config] get failover ok, dataId={}, group={}, tenant={}, config={}",
                        worker.getAgentName(), dataId, group, tenant, ContentUtils.truncateContent(content));
            cr.setContent(content);
            StringencryptedDataKey= LocalEncryptedDataKeyProcessor
                .getEncryptDataKeyFailover(agent.getName(), dataId, group, tenant);
            cr.setEncryptedDataKey(encryptedDataKey);
            configFilterChainManager.doFilter(null, cr);
            content = cr.getContent();
            return content;
        }
    
        try {
            // 通过客户端远程拉取配置信息
            ConfigResponseresponse= worker.getServerConfig(dataId, group, tenant, timeoutMs, false);
            cr.setContent(response.getContent());
            cr.setEncryptedDataKey(response.getEncryptedDataKey());
            configFilterChainManager.doFilter(null, cr);
            content = cr.getContent();
    
            return content;
        } catch (NacosException ioe) {
            if (NacosException.NO_RIGHT == ioe.getErrCode()) {
                throw ioe;
            }
            LOGGER.warn("[{}] [get-config] get from server error, dataId={}, group={}, tenant={}, msg={}",
                        worker.getAgentName(), dataId, group, tenant, ioe.toString());
        }
        // 非鉴权失败的异常的，可以从本地快照中获取配置，如果有的话
        content = LocalConfigInfoProcessor.getSnapshot(worker.getAgentName(), dataId, group, tenant);
        if (content != null) {
            LOGGER.warn("[{}] [get-config] get snapshot ok, dataId={}, group={}, tenant={}, config={}",
                        worker.getAgentName(), dataId, group, tenant, ContentUtils.truncateContent(content));
        }
        cr.setContent(content);
        StringencryptedDataKey= LocalEncryptedDataKeyProcessor
            .getEncryptDataKeySnapshot(agent.getName(), dataId, group, tenant);
        cr.setEncryptedDataKey(encryptedDataKey);
        configFilterChainManager.doFilter(null, cr);
        content = cr.getContent();
        return content;
    }
    复制代码

获取配置信息经历了三个步骤：

1.  从本地失败转移的文件夹中获取配置，这个是手工添加的，程序不会自动处理，是针对一些特定情况下，比如服务挂了还需要修改本地的配置的情况。给了一次本地修改处理的方式，也算是预留了一个备案，防止一些极端情况。但是这个得了解其固定的目录和拉取的配置，再处理，用完了要及时删除，否则会一直拉取本地的文件，毕竟是优先处理这部分的逻辑

1.  去服务端拉取，这个就是正常逻辑，获取服务端存储的配置信息

1.  对于出现了比如超时的情况，在有本地快照的情况的，从本地快照拉取配置，不至于偶尔超时了就配置没了

## 监听配置 ##

监听采用的回调的思想，当服务端通知的时候，调用回调方法。

@OverridepublicvoidaddListener(String dataId, String group, Listener listener)throws NacosException {
        worker.addTenantListeners(dataId, group, Arrays.asList(listener));
    }
    publicvoidaddTenantListeners(String dataId, String group, List<? extends Listener> listeners)
                throws NacosException {
        group = blank2defaultGroup(group);
        Stringtenant= agent.getTenant();
        // 根据dataId,group和listeners获取一个cacheData
        CacheDatacache= addCacheDataIfAbsent(dataId, group, tenant);
        synchronized (cache) {
            for (Listener listener : listeners) {
                cache.addListener(listener);
            }
            // 非丢弃，删除类型
            cache.setDiscard(false);
            // 未同步到服务端
            cache.setSyncWithServer(false);
            // 处理配置类
            agent.notifyListenConfig();
        }
    }
    
    privatefinal BlockingQueue<Object> listenExecutebell = newArrayBlockingQueue<>(1);
    @OverridepublicvoidnotifyListenConfig() {
        // listenExecutebell是一个阻塞队列，放入bellItem，即一个触发条件，相当于生产者
        listenExecutebell.offer(bellItem);
    }
    
    @OverridepublicvoidstartInternal() {
        // 线程池在阻塞等到信号的到来
        executor.schedule(() -> {
            while (!executor.isShutdown() && !executor.isTerminated()) {
                try {
                    // 获取到listenExecutebell.offer(bellItem)的信号
                    listenExecutebell.poll(5L, TimeUnit.SECONDS);
                    if (executor.isShutdown() || executor.isTerminated()) {
                        continue;
                    }
                    // 触发执行监听
                    executeConfigListen();
                } catch (Throwable e) {
                    LOGGER.error("[ rpc listen execute ] [rpc listen] exception", e);
                }
            }
        }, 0L, TimeUnit.MILLISECONDS);
    }
    
    @OverridepublicvoidexecuteConfigListen() {
        // 省略其他代码...
        for (CacheData cache : cacheMap.get().values()) {
            synchronized (cache) {
                //check local listeners consistent.
                if (cache.isSyncWithServer()) {
                    // 检查监听的md5值
                    cache.checkListenerMd5();
                    if (!needAllSync) {
                        continue;
                    }
                }
                // 省略其他代码...
            }
        }
        // 省略其他代码...
    }
    
    voidcheckListenerMd5() {
        for (ManagerListenerWrap wrap : listeners) {
            if (!md5.equals(wrap.lastCallMd5)) {
                // 通知变动的配置信息
                safeNotifyListener(dataId, group, content, type, md5, encryptedDataKey, wrap);
            }
        }
    }
    
    privatevoidsafeNotifyListener(final String dataId, final String group, final String content, final String type,
                final String md5, final String encryptedDataKey, final ManagerListenerWrap listenerWrap) {
         // 省略其他代码...
         Runnablejob= () -> {
             // 省略其他代码...
             // 触发回调
             listener.receiveConfigInfo(contentTmp);
             // 省略其他代码...
         };
         finallongstartNotify= System.currentTimeMillis();
         try {
             // 监听器配置了异步执行器，就用配置的执行
             if (null != listener.getExecutor()) {
                 listener.getExecutor().execute(job);
             } else {
                 try {
                     // 内部线程池执行
                     INTERNAL_NOTIFIER.submit(job);
                 } catch (RejectedExecutionException rejectedExecutionException) {
                     // 省略其他代码...
                     job.run();
                 } catch (Throwable throwable) {
                     // 省略其他代码...
                     job.run();
                 }
             }
         } catch (Throwable t) {
            // 省略其他代码...
         }
         // 省略其他代码...
     }   
    复制代码

监听配置是在cacheData中配置上监听器，等待触发条件后，进行本地的内容和远程内容的比对，如果不一致，调用监听器上的回调逻辑，完成配置的更新通知。

## Spring-Cloud-alibaba-nacos-config的处理 ##

经过对以上的分析，我们再看一看Spring-Cloud-alibaba-nacos-config配置的代码是如何完成配置的更新的(这里用的是spring-cloud-alibaba 2.2.0.RELEASE版本的代码)。

publicclassNacosConfigAutoConfiguration {
        // 省略其他代码...
        @Beanpublic NacosContextRefresher nacosContextRefresher(
                NacosConfigManager nacosConfigManager,
                NacosRefreshHistory nacosRefreshHistory) {
            // 创建了一个nacos上下文刷新器returnnewNacosContextRefresher(nacosConfigManager, nacosRefreshHistory);
        }
    }
    
    publicclassNacosContextRefresherimplementsApplicationListener<ApplicationReadyEvent>, ApplicationContextAware {
        // 省略其他代码...
        @OverridepublicvoidonApplicationEvent(ApplicationReadyEvent event) {
            if (this.ready.compareAndSet(false, true)) {
                 // 注册nacos监听this.registerNacosListenersForApplications();
            }
        }
        // 省略其他代码...
        privatevoidregisterNacosListenersForApplications() {
            if (isRefreshEnabled()) {
                for (NacosPropertySource propertySource : NacosPropertySourceRepository
                     .getAll()) {
                    if (!propertySource.isRefreshable()) {
                        continue;
                    }
                    StringdataId= propertySource.getDataId();
                    // 注册nacos监听
                    registerNacosListener(propertySource.getGroup(), dataId);
                }
            }
        }
        
        privatevoidregisterNacosListener(final String groupKey, final String dataKey) {
            Stringkey= NacosPropertySourceRepository.getMapKey(dataKey, groupKey);
             // 创建监听 Listenerlistener= listenerMap.computeIfAbsent(key,
                lst -> newAbstractSharedListener() {
                    @OverridepublicvoidinnerReceive(String dataId, String group,
                            String configInfo) {
                        refreshCountIncrement();
                        nacosRefreshHistory.addRefreshRecord(dataId, group, configInfo);
                        // 当有配置信息变动的时候，通知spring cloud 刷新事件
                        applicationContext.publishEvent(
                                newRefreshEvent(this, null, "Refresh Nacos config"));
                        if (log.isDebugEnabled()) {
                            log.debug(String.format(
                                    "Refresh Nacos config group=%s,dataId=%s,configInfo=%s",
                                    group, dataId, configInfo));
                        }
                    }
                });
            try {
                // 很熟悉，对不对，绑定监听
                configService.addListener(dataKey, groupKey, listener);
            }
            catch (NacosException e) {
                log.warn(String.format(
                        "register fail for nacos listener ,dataId=[%s],group=[%s]", dataKey,
                        groupKey), e);
            }
        }
    }
    复制代码

其代码主要逻辑如下：

1.  在spring.factories中配置自动配置类NacosConfigAutoConfiguration

1.  NacosConfigAutoConfiguration创建NacosContextRefresher

1.  NacosContextRefresher监听ApplicationReadyEvent事件

1.  触发ApplicationReadyEvent事件后注册监听，创建监听配置的dataId，group对应的配置信息的监听类

1.  configService绑定监听器，如果监听到配置信息的变化，则发布RefreshEvent

1.  spring cloud监听到RefreshEvent后刷新配置，刷新spring容器

## 总结 ##

本篇从ConfigExample入手，主要介绍了配置服务是如何初始化，获取配置信息的。并且分析了监听器的作用已经简单分析了spring-cloud-alibaba-nacos-config的部分源码，了解了nacos是如何结合spring-cloud。下篇将重点介绍ClientWorker和cacheData。敬请期待。

作者：ruipost

链接：[https://juejin.cn/post/7212937418960683065][https_juejin.cn_post_7212937418960683065]

[https_juejin.cn_post_7212937418960683065]: https://juejin.cn/post/7212937418960683065