从Nacos客户端视角来分析一下配置中心实现原理-网络知识

一动态配置
- 1. 环境准备
- 2.新建配置
- 3.导入配置
- 4.配置客户端
- 5. 修改配置信息
- 6.小结
二配置中心原理(推还是拉)
- 1.实例化 ConfigService
- 2.添加 Listener
- 3.CacheData
- 4.触发回调
- 5.Md5何时变更
- 6.拉的优势
三总结

Hello,大家好,我是麦洛,今天我们一起从Nacos客户端视角来看看配置中心实现原理;整理这篇文章时候,也参照学习了部分大佬的博客,这里致谢;大家在阅读过程中如果发现错误或者问题,可以在留言板留言,也可以后台回复"加群"获取我的微信二维码,加我一起交流学习;

在开始阅读文章之前,有些思路我按我的理解先阐述一些,方便大家更快理清思路,不对的地方还请大家批评指正;

Nacos客户端会在在本地缓存服务端配置文件,防止服务器奔溃情况下,导致服务不可用;
本地缓存类在代码中的体现就是我们下面提到的CacheData,我们知道对应服务端一个配置,肯定可以同时被多个客户端所使用,当这个配置发生变更,如何去通知到每一个客户端?
客户端启动之后,回去注册监视器,监视器最终会被保存到CacheData类中CopyOnWriteArrayList<ManagerListenerWrap> listeners字段,那么,反过来,当执行监视器回调方法时,就可以找到所有客户端
长轮询左右主要就是刷新配置,保持服务端配置和本地缓存配置保持一致;

首先,我们来看看Nacos官网给出的Nacos地图,我们可以清楚的看到,动态配置服务是 Nacos 的三大功能之一;

这里借用官网的描述,一起来看看Nacos 为我们带来什么黑科技?

动态配置服务可以让您以中心化、外部化和动态化的方式管理所有环境的应用配置和服务配置。动态配置消除了配置变更时重新部署应用和服务的需要，让配置管理变得更加高效和敏捷。配置中心化管理让实现无状态服务变得更简单，让服务按需弹性扩展变得更容易。

所以,有了Nacos ,可能我们以前上线打包弄错配置文件,改配置需要重启服务等一系列问题,都会显著改观

一动态配置

下面我将来和大家一起来了解下 Nacos 的动态配置的能力，看看 Nacos 是如何以简单、优雅、高效的方式管理配置，实现配置的动态变更的。

我们用一个简单的例子来了解下 Nacos 的动态配置的功能。

1. 环境准备

首先,我们需要搭建一个Nacos 服务端,由于官网的quick-start已经对此做了详细的解读,我们这里就不在赘述

https://nacos.io/zh-cn/docs/quick-start.html

安装完成之后启动，我们就可以访问 Nacos 的控制台了，如下图所示：

Nacos控制台做了简单的权限控制，默认的账号和密码都是 nacos。

登录进去之后，是这样的：

2.新建配置

接下来我们在控制台上创建一个简单的配置项，如下图所示：

3.导入配置

Nacos支持导入配置,可以直接将配置文件压缩包导入,这里我们以人人开源的微服务项目为例

4.配置客户端

下面我以自己搭建的子服务为例,一起来看看Nacos配置中心的使用

首先我们需要配置一下,大家只需关注config节点配置就可以,discovery节点可以忽略

  cloud:
    nacos:
      discovery:
        metadata:
          management:
            context-path: ${server.servlet.context-path}/actuator
        server-addr: ${nacos-host:nacos-host}:${nacos-port:8848}
        #nacos的命名空间ID，默认是public
        namespace: ${nacos-namespace:}
        service: ets-web
      config:
        server-addr: ${spring.cloud.nacos.discovery.server-addr}
        namespace: ${spring.cloud.nacos.discovery.namespace}
        group: RENREN_CLOUD_GROUP
        file-extension: yaml
        #指定共享配置，且支持动态刷新
        extension-configs:
          - data-id: datasource.yaml
            group: ${spring.cloud.nacos.config.group}
            refresh: true
          - data-id: common.yaml
            group: ${spring.cloud.nacos.config.group}
            refresh: true

其实extension-configs节点的配置信息对应的是下面的类

接下来我们启动服务,来看看控制台日志

5. 修改配置信息

接下来我们在 Nacos 的控制台上将我们的配置信息改为如下图所示：

修改完配置，点击 “发布” 按钮后，客户端将会收到最新的数据，如下图所示：

至此一个简单的动态配置管理功能已经讲完了，删除配置和更新配置操作类似，这里不再赘述。

6.小结

通过上面的小案例,我们大概了解了Nacos动态配置的服务的使用方法,Nacos服务端将配置信息保存到其配置文件所配置的数据库中,客户端连接到服务端之后,根据 dataID，Group可以获取到具体的配置信息，当服务端的配置发生变更时，客户端会收到通知。当客户端拿到变更后的最新配置信息后，就可以做自己的处理了，这非常有用，所有需要使用配置的场景都可以通过 Nacos 来进行管理。

二配置中心原理(推还是拉)

现在我们了解了 Nacos 的动态配置服务的功能了，但是有一个问题我们需要弄明白，那就是 Nacos 客户端是怎么实时获取到 Nacos 服务端的最新数据的。

其实客户端和服务端之间的数据交互，无外乎两种情况：

服务端推数据给客户端
客户端从服务端拉数据

那到底是推还是拉呢，从 Nacos 客户端通过 Listener 来接收最新数据的这个做法来看，感觉像是服务端推的数据，但是不能想当然，要想知道答案，最快最准确的方法就是从源码中去寻找。

官方示例代码

try {
    // 传递配置
    String serverAddr = "{serverAddr}";
    String dataId = "{dataId}";
    String group = "{group}";
    Properties properties = new Properties();
    properties.put("serverAddr", serverAddr);

    // 新建 configService
    ConfigService configService = NacosFactory.createConfigService(properties);
    String content = configService.getConfig(dataId, group, 5000);
    System.out.println(content);

    // 注册监听器
    configService.addListener(dataId, group, new Listener() {
    @Override
    public void receiveConfigInfo(String configInfo) {
        System.out.println("recieve1:" + configInfo);
    }
    @Override
    public Executor getExecutor() {
        return null;
    }
});
} catch (NacosException e) {
    // TODO 
    -generated catch block
    e.printStackTrace();
}

1.实例化 ConfigService

当我们引包结束以后,会发现下面三个关于Nacos的包

从我的理解来说,api包会调用client包的能力来和Nacos服务端进行交互.那再交互时候,主要就会用到我们接下来分析的实现了ConfigService接口的NacosConfigService 类

现在我们来看下 NacosConfigService 的构造方法，看看 ConfigService 是怎么实例化的，如下图所示：

public class NacosConfigService implements ConfigService {
    
    private static final Logger LOGGER = LogUtils.logger(NacosConfigService.class);
    
    private static final long POST_TIMEOUT = 3000L;
    
    /**
     * http agent.
     */
    private final HttpAgent agent;
    
    /**
     * long polling.  这里是长轮询
     */
    private final ClientWorker worker;
    
    private String namespace;
    
    private final String encode;
    //省略其他代码

    //构造方法
public NacosConfigService(Properties properties) throws NacosException {
        ValidatorUtils.checkInitParam(properties);
        String encodeTmp = properties.getProperty(PropertyKeyConst.ENCODE);
        if (StringUtils.isBlank(encodeTmp)) {
            this.encode = Constants.ENCODE;
        } else {
            this.encode = encodeTmp.trim();
        }
        initNamespace(properties);
        //对象1
        this.agent = new MetricsHttpAgent(new ServerHttpAgent(properties));
        this.agent.start();
        //对象2
        this.worker = new ClientWorker(this.agent, this.configFilterChainManager, properties);
    }

实例化时主要是初始化了两个对象，他们分别是：

HttpAgent
ClientWorker

HttpAgent

其中 agent 是通过装饰器模式实现的，ServerHttpAgent 是实际工作的类，MetricsHttpAgent 在内部也是调用了 ServerHttpAgent 的方法，另外加上了一些统计操作，所以我们只需要关心 ServerHttpAgent 的功能就可以了。

不熟悉的同学,可以看菜鸟教程对装饰器模式的解读

agent 实际是在 ClientWorker 中发挥能力的，而 ClientWorker 也是真正的打工人,下面我们来看下 ClientWorker 类。

ClientWorker

以下是 ClientWorker 的构造方法，如下图所示：

    public ClientWorker(final HttpAgent agent, final ConfigFilterChainManager configFilterChainManager,
            final Properties properties) {
        this.agent = agent;
        this.configFilterChainManager = configFilterChainManager;
        
        // Initialize the timeout parameter
        
        init(properties);
        //创建了一个定时任务的线程池
        this.executor = Executors.newScheduledThreadPool(1, new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r);
                t.setName("com.alibaba.nacos.client.Worker." + agent.getName());
                t.setDaemon(true);
                return t;
            }
        });
        //创建了一个保持长轮询的线程池
        this.executorService = Executors
                .newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() {
                    @Override
                    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;
                    }
                });
        
        //创建了一个延迟任务线程池来每隔10ms来检查配置信息的线程池
        this.executor.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                try {
                    checkConfigInfo();
                } catch (Throwable e) {
                    LOGGER.error("[" + agent.getName() + "] [sub-check] rotate check error", e);
                }
            }
        }, 1L, 10L, TimeUnit.MILLISECONDS);
    }

可以看到 ClientWorker 除了将 HttpAgent 维持在自己内部，还创建了两个线程池：

final ScheduledExecutorService executor;
    
final ScheduledExecutorService executorService;

第一个线程池负责与配置中心进行数据的交互，并且启动后延迟1ms，之后每隔10ms对配置信息进行定时检查
第二个线程池则是负责保持一个长轮询链接

接下来让我们来看下 executor 每 10ms 执行的方法到底做了什么工作，如下图所示：

    /**
     * groupKey -> cacheData.
     */
    private final AtomicReference<Map<String, CacheData>> cacheMap = new AtomicReference<Map<String, CacheData>>(
            new HashMap<String, CacheData>());

    /**
     * Check config info. 检查配置信息
     */
    public void checkConfigInfo() {
        // 分任务（解决大数据量的传输问题）
        int listenerSize = cacheMap.get().size();
        // 向上取整为批数，分批次进行检查
        //ParamUtil.getPerTaskConfigSize() =3000
        int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());
        if (longingTaskCount > currentLongingTaskCount) {
            for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {
                // 要判断任务是否在执行 这块需要好好想想。 任务列表现在是无序的。变化过程可能有问题
                executorService.execute(new LongPollingRunnable(i));
            }
            currentLongingTaskCount = longingTaskCount;
        }
    }

这里主要是先去拿缓存中 Map<String, CacheData> 的数量，为避免处理过量的数据，这里对缓存数据进行了分组 ,然后提交给 executorService 线程池去执行，执行的任务就是 LongPollingRunnable，每个任务都有一个 taskId。

现在我们来看看 LongPollingRunnable 做了什么，主要分为两部分，

第一部分是检查本地的配置信息，
第二部分是获取服务端的配置信息然后更新到本地。

1.本地检查

首先取出与该 taskId 相关的 CacheData，然后对 CacheData 进行检查，包括本地配置检查和缓存数据的 md5 检查，本地检查主要是做一个故障容错，当服务端挂掉后，Nacos 客户端可以从本地的文件系统中获取相关的配置信息，如下图所示：

public void run() {
            
            List<CacheData> cacheDatas = new ArrayList<CacheData>();
            List<String> inInitializingCacheList = new ArrayList<String>();
            try {
                //
                for (CacheData cacheData : cacheMap.get().values()) {
                    if (cacheData.getTaskId() == taskId) {
                        cacheDatas.add(cacheData);
                        try {
                            //执行检查本地配置
                            checkLocalConfig(cacheData);
                            if (cacheData.isUseLocalConfigInfo()) {
                                //缓存数据的md5的检查
                                cacheData.checkListenerMd5();
                            }
                        } catch (Exception e) {
                            LOGGER.error("get local config info error", e);
                        }
                    }
                }
              
        }

    
//检查本地配置
    
private void checkLocalConfig(CacheData cacheData) {
        final String dataId = cacheData.dataId;
        final String group = cacheData.group;
        final String tenant = cacheData.tenant;
    //本地缓存文件
        File path = LocalConfigInfoProcessor.getFailoverFile(agent.getName(), dataId, group, tenant);
        //不使用本地配置，但是持久化文件存在，需要读取文件加载至内存
        if (!cacheData.isUseLocalConfigInfo() && path.exists()) {
            String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
            final String md5 = MD5Utils.md5Hex(content, Constants.ENCODE);
            cacheData.setUseLocalConfigInfo(true);
            cacheData.setLocalConfigInfoVersion(path.lastModified());
            cacheData.setContent(content);
            
            LOGGER.warn(
                    "[{}] [failover-change] failover file created. dataId={}, group={}, tenant={}, md5={}, content={}",
                    agent.getName(), dataId, group, tenant, md5, ContentUtils.truncateContent(content));
            return;
        }
        
       // 有 -> 没有。不通知业务监听器，从server拿到配置后通知。
        //使用本地配置，但是持久化文件不存在 
        if (cacheData.isUseLocalConfigInfo() && !path.exists()) {
            cacheData.setUseLocalConfigInfo(false);
            LOGGER.warn("[{}] [failover-change] failover file deleted. dataId={}, group={}, tenant={}", agent.getName(),
                    dataId, group, tenant);
            return;
        }
        
        // 有变更
        //使用本地配置，持久化文件存在，缓存跟文件最后修改时间不一致
        if (cacheData.isUseLocalConfigInfo() && path.exists() && cacheData.getLocalConfigInfoVersion() != path
                .lastModified()) {
            String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
            final String md5 = MD5Utils.md5Hex(content, Constants.ENCODE);
            cacheData.setUseLocalConfigInfo(true);
            cacheData.setLocalConfigInfoVersion(path.lastModified());
            cacheData.setContent(content);
            LOGGER.warn(
                    "[{}] [failover-change] failover file changed. dataId={}, group={}, tenant={}, md5={}, content={}",
                    agent.getName(), dataId, group, tenant, md5, ContentUtils.truncateContent(content));
        }
    }

本地检查主要是通过是否使用本地配置，继而寻找持久化缓存文件，再通过判断文件的最后修改事件与本地缓存的版本是否一致来判断是否由变更

通过跟踪 checkLocalConfig 方法，可以看到 Nacos 将缓存配置信息保存在了

~/nacos/config/fixed-{address}_8848_nacos/snapshot/DEFAULT_GROUP/{dataId}

这个文件中，我们看下这个文件中保存的内容，如下图所示：

2.服务端检查

然后通过 checkUpdateDataIds() 方法从服务端获取值变化的 dataId 列表，

通过 getServerConfig 方法，根据 dataId 到服务端获取最新的配置信息，接着将最新的配置信息保存到 CacheData 中。

最后调用 CacheData 的 checkListenerMd5 方法，可以看到该方法在第一部分也被调用过，我们需要重点关注一下。

                // 检查服务器配置
                List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);
                if (!CollectionUtils.isEmpty(changedGroupKeys)) {
                    LOGGER.info("get changedGroupKeys:" + changedGroupKeys);
                }
                
                for (String groupKey : changedGroupKeys) {
                    String[] key = GroupKey.parseKey(groupKey);
                    String dataId = key[0];
                    String group = key[1];
                    String tenant = null;
                    if (key.length == 3) {
                        tenant = key[2];
                    }
                    try {
                        //从服务器端获取相关id的最新配置
                        String[] ct = getServerConfig(dataId, group, tenant, 3000L);
                        CacheData cache = cacheMap.get().get(GroupKey.getKeyTenant(dataId, group, tenant));
                        cache.setContent(ct[0]);
                        if (null != ct[1]) {
                            cache.setType(ct[1]);
                        }
                        LOGGER.info("[{}] [data-received] dataId={}, group={}, tenant={}, md5={}, content={}, type={}",
                                agent.getName(), dataId, group, tenant, cache.getMd5(),
                                ContentUtils.truncateContent(ct[0]), ct[1]);
                    } catch (NacosException ioe) {
                        String message = String
                                .format("[%s] [get-update] get changed config exception. dataId=%s, group=%s, tenant=%s",
                                        agent.getName(), dataId, group, tenant);
                        LOGGER.error(message, ioe);
                    }
                }
                for (CacheData cacheData : cacheDatas) {
                    if (!cacheData.isInitializing() || inInitializingCacheList
                            .contains(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant))) {
                        //校验MD5值
                        cacheData.checkListenerMd5();
                        cacheData.setInitializing(false);
                    }
                }
                inInitializingCacheList.clear();
                
                executorService.execute(this);
                
            } catch (Throwable e) {
                
                // If the rotation training task is abnormal, the next execution time of the task will be punished
                LOGGER.error("longPolling error : ", e);
                executorService.schedule(this, taskPenaltyTime, TimeUnit.MILLISECONDS);
            }

这里大家也发现,当客户端从服务器拉去配置文件之后,会将配置文件在本地进行缓存,所以,一般会优先使用本地配置,如果本地文件不存在或者内容为空，则再通过 HTTP GET 方法从远端拉取配置，并保存到本地缓存中

    private String getConfigInner(String tenant, String dataId, String group, long timeoutMs) throws NacosException {
        group = null2defaultGroup(group);
        ParamUtils.checkKeyParam(dataId, group);
        ConfigResponse cr = new ConfigResponse();
        
        cr.setDataId(dataId);
        cr.setTenant(tenant);
        cr.setGroup(group);
        
        // 优先使用本地配置
        String content = LocalConfigInfoProcessor.getFailover(agent.getName(), dataId, group, tenant);
        if (content != null) {
            LOGGER.warn("[{}] [get-config] get failover ok, dataId={}, group={}, tenant={}, config={}", agent.getName(),
                    dataId, group, tenant, ContentUtils.truncateContent(content));
            cr.setContent(content);
            configFilterChainManager.doFilter(null, cr);
            content = cr.getContent();
            return content;
        }
        
        try {
            String[] ct = worker.getServerConfig(dataId, group, tenant, timeoutMs);
            cr.setContent(ct[0]);
            
            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={}",
                    agent.getName(), dataId, group, tenant, ioe.toString());
        }
        
        LOGGER.warn("[{}] [get-config] get snapshot ok, dataId={}, group={}, tenant={}, config={}", agent.getName(),
                dataId, group, tenant, ContentUtils.truncateContent(content));
        content = LocalConfigInfoProcessor.getSnapshot(agent.getName(), dataId, group, tenant);
        cr.setContent(content);
        configFilterChainManager.doFilter(null, cr);
        content = cr.getContent();
        return content;
    }

2.添加 Listener

好了现在我们可以为 ConfigService 来添加一个 Listener 了，最终是调用了 ClientWorker 的 addTenantListeners 方法，如下图所示：

    /**
     * Add listeners for tenant.
     *
     * @param dataId    dataId of data
     * @param group     group of data
     * @param listeners listeners
     * @throws NacosException nacos exception
     */
    public void addTenantListeners(String dataId, String group, List<? extends Listener> listeners)
            throws NacosException {
        //设置默认组
        group = null2defaultGroup(group);
        String tenant = agent.getTenant();
        CacheData cache = addCacheDataIfAbsent(dataId, group, tenant);
        for (Listener listener : listeners) {
            cache.addListener(listener);
        }
    }

该方法分为两个部分，首先根据 dataId，group 和tenant获取一个 CacheData 对象，然后将当前要添加的 listener 对象添加到 CacheData 中去。

接下来,我们要重点关注下 CacheData 类了。

3.本地缓存CacheData

首先让我们来看一下 CacheData 中的成员变量，如下图所示：

    
    private final String name;
    
    private final ConfigFilterChainManager configFilterChainManager;
    
    public final String dataId;
    
    public final String group;
    
    public final String tenant;
    //监听器
    
    private final CopyOnWriteArrayList<ManagerListenerWrap> listeners;
    
    private volatile String md5;
    
    /**
     * whether use local config.
     */
    private volatile boolean isUseLocalConfig = false;
    
    /**
     * last modify time.
     */
    private volatile long localConfigLastModified;
    
    private volatile String content;
    
    private int taskId;
    
    private volatile boolean isInitializing = true;
    
    private String type;

我们可以看到,成员变量包括tenant ,dataId，group，content，taskId等,还有两个值得我们关注的:

listeners
md5

listeners 是该 CacheData 所关联的所有 listener，不过不是保存的原始的 Listener 对象，而是包装后的 ManagerListenerWrap 对象，该对象除了持有 Listener 对象，还持有了一个 lastCallMd5 和lastContent属性。

    private static class ManagerListenerWrap {
        
        final Listener listener;
        
        //关注
        String lastCallMd5 = CacheData.getMd5String(null);
        
        String lastContent = null;
        
        ManagerListenerWrap(Listener listener) {
            this.listener = listener;
        }
        
        ManagerListenerWrap(Listener listener, String md5) {
            this.listener = listener;
            this.lastCallMd5 = md5;
        }
        
        ManagerListenerWrap(Listener listener, String md5, String lastContent) {
            this.listener = listener;
            this.lastCallMd5 = md5;
            this.lastContent = lastContent;
        }
        
    }

另外一个属性 md5 就是根据当前对象的 content 计算出来的 md5 值。

4.触发监听器回调

现在我们对 ConfigService 有了大致的了解了，现在剩下最后一个重要的问题还没有答案，那就是 ConfigService 的 Listener 是在什么时候触发回调方法 receiveConfigInfo 的。

现在让我们回过头来想一下，在 ClientWorker 中的定时任务中，启动了一个长轮询的任务：LongPollingRunnable，该任务多次执行了 cacheData.checkListenerMd5() 方法，那现在就让我们来看下这个方法到底做了些什么，如下图所示：

    
    void checkListenerMd5() {
        for (ManagerListenerWrap wrap : listeners) {
            if (!md5.equals(wrap.lastCallMd5)) {
                safeNotifyListener(dataId, group, content, type, md5, wrap);
            }
        }
    }

到这里应该就比较清晰了，该方法会检查 CacheData 当前的 md5 与 CacheData 持有的所有 Listener 中保存的 md5 的值是否一致，如果不一致，就执行一个安全的监听器的通知方法：safeNotifyListener，通知什么呢？我们可以大胆的猜一下，应该是通知 Listener 的使用者，该 Listener 所关注的配置信息已经发生改变了。现在让我们来看一下 safeNotifyListener 方法，如下图所示：

    private void safeNotifyListener(final String dataId, final String group, final String content, final String type,
            final String md5, final ManagerListenerWrap listenerWrap) {
        final Listener listener = listenerWrap.listener;
        
        Runnable job = new Runnable() {
            @Override
            public void run() {
                ClassLoader myClassLoader = Thread.currentThread().getContextClassLoader();
                ClassLoader appClassLoader = listener.getClass().getClassLoader();
                try {
                    if (listener instanceof AbstractSharedListener) {
                        AbstractSharedListener adapter = (AbstractSharedListener) listener;
                        adapter.fillContext(dataId, group);
                        LOGGER.info("[{}] [notify-context] dataId={}, group={}, md5={}", name, dataId, group, md5);
                    }
                    // 执行回调之前先将线程classloader设置为具体webapp的classloader，以免回调方法中调用spi接口是出现异常或错用（多应用部署才会有该问题）。
                    Thread.currentThread().setContextClassLoader(appClassLoader);
                    
                    ConfigResponse cr = new ConfigResponse();
                    cr.setDataId(dataId);
                    cr.setGroup(group);
                    cr.setContent(content);
                    
                    //重点关注,在这里调用
                    //重点关注,在这里调用
                    //重点关注,在这里调用
                    
                    configFilterChainManager.doFilter(null, cr);
                    String contentTmp = cr.getContent();
                    listener.receiveConfigInfo(contentTmp);
                    
                    
                    
                    
                    // compare lastContent and content
                    if (listener instanceof AbstractConfigChangeListener) {
                        Map data = ConfigChangeHandler.getInstance()
                                .parseChangeData(listenerWrap.lastContent, content, type);
                        ConfigChangeEvent event = new ConfigChangeEvent(data);
                        ((AbstractConfigChangeListener) listener).receiveConfigChange(event);
                        listenerWrap.lastContent = content;
                    }
                    
                    listenerWrap.lastCallMd5 = md5;
                    LOGGER.info("[{}] [notify-ok] dataId={}, group={}, md5={}, listener={} ", name, dataId, group, md5,
                            listener);
                } catch (NacosException ex) {
                    LOGGER.error("[{}] [notify-error] dataId={}, group={}, md5={}, listener={} errCode={} errMsg={}",
                            name, dataId, group, md5, listener, ex.getErrCode(), ex.getErrMsg());
                } catch (Throwable t) {
                    LOGGER.error("[{}] [notify-error] dataId={}, group={}, md5={}, listener={} tx={}", name, dataId,
                            group, md5, listener, t.getCause());
                } finally {
                    Thread.currentThread().setContextClassLoader(myClassLoader);
                }
            }
        };
        
        final long startNotify = System.currentTimeMillis();
        try {
            if (null != listener.getExecutor()) {
                listener.getExecutor().execute(job);
            } else {
                job.run();
            }
        } catch (Throwable t) {
            LOGGER.error("[{}] [notify-error] dataId={}, group={}, md5={}, listener={} throwable={}", name, dataId,
                    group, md5, listener, t.getCause());
        }
        final long finishNotify = System.currentTimeMillis();
        LOGGER.info("[{}] [notify-listener] time cost={}ms in ClientWorker, dataId={}, group={}, md5={}, listener={} ",
                name, (finishNotify - startNotify), dataId, group, md5, listener);
    }

可以看到在 safeNotifyListener 方法中，重点关注下红框中的三行代码：获取最新的配置信息，调用 Listener 的回调方法，将最新的配置信息作为参数传入，这样 Listener 的使用者就能接收到变更后的配置信息了，最后更新 ListenerWrap 的 md5 值。和我们猜测的一样， Listener 的回调方法就是在该方法中触发的。

5.Md5何时变更

那 CacheData 的 md5 值是何时发生改变的呢？我们可以回想一下，在上面的 LongPollingRunnable 所执行的任务中，在获取服务端发生变更的配置信息时，将最新的 content 数据写入了 CacheData 中，我们可以看下该方法如下：

    public void setContent(String content) {
        this.content = content;
        this.md5 = getMd5String(this.content);
    }

可以看到是在长轮询的任务中，当服务端配置信息发生变更时，客户端将最新的数据获取下来之后，保存在了 CacheData 中，同时更新了该 CacheData 的 md5 值，所以当下次执行 checkListenerMd5 方法时，就会发现当前 listener 所持有的 md5 值已经和 CacheData 的 md5 值不一样了，也就意味着服务端的配置信息发生改变了，这时就需要将最新的数据通知给 Listener 的持有者。

至此配置中心的完整流程已经分析完毕了，可以发现，Nacos 并不是通过推的方式将服务端最新的配置信息发送给客户端的，而是客户端维护了一个长轮询的任务，定时去拉取发生变更的配置信息，然后将最新的数据推送给 Listener 的持有者。

6.为什么要拉?

客户端拉取服务端的数据与服务端推送数据给客户端相比，优势在哪呢，为什么 Nacos 不设计成主动推送数据，而是要客户端去拉取呢？如果用推的方式，服务端需要维持与客户端的长连接，这样的话需要耗费大量的资源，并且还需要考虑连接的有效性，例如需要通过心跳来维持两者之间的连接。而用拉取的方式，客户端只需要通过一个无状态的 http 请求即可获取到服务端的数据。

三总结

现在,我们来简单复盘一下Nacos客户端视角下的配置中心实现原理

首先我们假设Nacos服务端一切正常,Nacos客户端启动以后

第一步是根据我们配置的服务端信息，新建 ConfigService 实例,它的实现就是我们文中提到的NacosConfigService;

第二步可以通过相应的接口获取配置和注册配置监听器,

考虑到服务端故障的问题，客户端将最新数据获取后会保存在本地的缓存文件中，以后会优先从文件中获取配置信息的值,如果获取不到,会直接从服务器拉去,并保存到缓存中;

其实真正干活的就是ClientWorker类;客户端是通过一个定时的长轮询来检查自己监听的配置项的数据的，一旦服务端的数据发生变化时，会从服务端获取到dataID的列表,

客户端根据dataID列表从服务端获取到最新的数据，并将最新的数据保存在一个 CacheData 对象中，在轮询过程中,如果决定使用本地配置,就会比较当前CacheData 的MD5值是否和所有监听者所持有的MD5值相等,如果不相等,,此时就会对该 CacheData 所绑定的 Listener 触发 receiveConfigInfo 回调,来通知使用者此配置信息已经变更;

谢谢大家的阅读,如果有什么疑问或者建议,可以点击下方留言板,给我留言!

一 动态配置

1. 环境准备

2.新建配置

3.导入配置

4.配置客户端

5. 修改配置信息

6.小结

二 配置中心原理(推还是拉)

1.实例化 ConfigService

HttpAgent

ClientWorker

2.添加 Listener

3.本地缓存CacheData

4.触发监听器回调

5.Md5何时变更

6.为什么要拉?

三 总结

相关文章

央媒谈“文盲演员”引发饭圈甩锅大战

野生大熊猫深夜遛达 镜头前呆萌打卡

张萌谈演员到底需不需要有文化

情人节多地迎领证高峰

婚房烂尾10年业主住进毛坯房

多地首套房贷利率降至4%以下 为什么急于提前还贷

一动态配置

二配置中心原理(推还是拉)

三总结

野生大熊猫深夜遛达镜头前呆萌打卡

多地首套房贷利率降至4%以下为什么急于提前还贷