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

电玩女神 2023-09-25 18:54 167阅读 0赞

## CacheData ##

首先需要讲解到的就是CacheData。所有的配置信息和监听器都存放在CacheData中，包括一些操作方法。所以它不仅仅是一个存储的配置信息的实体，也包括了操作，获取，更新这个信息的功能。以下通过代码分析一下它的私有变量。

publicclassCacheData {
        // 线程池的最大线程数    staticfinalintCONCURRENCY=5;
        // 线程工厂，表明创建的是后台线程并赋予名字  staticThreadFactoryinternalNotifierFactory= r -> {
            Threadt=newThread(r);
            t.setName("nacos.client.cachedata.internal.notifier");
            t.setDaemon(true);
            return t;
        };
        // 是否初始化快照staticboolean initSnapshot;
        // 静态块，获取属性值static {
            initSnapshot = NacosClientProperties.PROTOTYPE.getBoolean("nacos.cache.data.init.snapshot", true);
            LOGGER.info("nacos.cache.data.init.snapshot = {} ", initSnapshot);
        }
        // 内部的通知线程池staticfinalThreadPoolExecutorINTERNAL_NOTIFIER=newThreadPoolExecutor(0, CONCURRENCY, 60L, TimeUnit.SECONDS,
                newSynchronousQueue<>(), internalNotifierFactory);
        // 名称privatefinal String name;
        // 过滤器链privatefinal ConfigFilterChainManager configFilterChainManager;
        // dataIdpublicfinal String dataId;
        // grouppublicfinal String group;
        // tenantpublicfinal String tenant;
        // 监听器privatefinal CopyOnWriteArrayList<ManagerListenerWrap> listeners;
        // 配置信息的md5值privatevolatile String md5;
        // 是否使用本地缓存privatevolatilebooleanisUseLocalConfig=false;
        // 上次配置修改时间，long类型privatevolatilelong localConfigLastModified;
        // 配置信息privatevolatile String content;
        // 加密数据的keyprivatevolatile String encryptedDataKey;
        // 修改时间privatevolatileAtomicLonglastModifiedTs=newAtomicLong(0);
        // taskIdprivateint taskId;
        // 是否初始化privatevolatilebooleanisInitializing=true;
        // 是否和服务端同步privatevolatilebooleanisSyncWithServer=false;
        // 是否丢失privatevolatilebooleanisDiscard=false;
        // typeprivate String type;
    }
    复制代码

CacheData的一些方法和成员变量的解释说明会在调用的时候再具体说明，这里只是需要有个印象，它不仅仅是对配置信息的保存，也包含了一些其他的状态位和操作的方法和变量。

## ClientWorker ##

对CacheData有了初步印象后，我们就开始重点分析一下ClientWorker。先看它的构造方法。

publicClientWorker(final ConfigFilterChainManager configFilterChainManager, ServerListManager serverListManager,
                final NacosClientProperties properties)throws NacosException {
        this.configFilterChainManager = configFilterChainManager;
        init(properties);
        // 创建一个用于配置服务端的Rpc通信客户端
        agent = newConfigRpcTransportClient(properties, serverListManager);
        intcount= ThreadUtils.getSuitableThreadCount(THREAD_MULTIPLE);
        ScheduledExecutorServiceexecutorService= Executors
            .newScheduledThreadPool(Math.max(count, MIN_THREAD_NUM), r -> {
                Threadt=newThread(r);
                t.setName("com.alibaba.nacos.client.Worker");
                t.setDaemon(true);
                return t;
            });
        // 配置了一个异步处理线程池
        agent.setExecutor(executorService);
        // 调用start方法
        agent.start();
    }
    
    publicvoidstart()throws NacosException {
        // 认证服务，主要是通过Secret Key和Access Key做认证用
        securityProxy.login(this.properties);
        this.executor.scheduleWithFixedDelay(() -> securityProxy.login(properties), 0,
                                             this.securityInfoRefreshIntervalMills, TimeUnit.MILLISECONDS);
        // 内部启动
        startInternal();
    }
    
    publicvoidstartInternal() {
        executor.schedule(() -> {
            while (!executor.isShutdown() && !executor.isTerminated()) {
                try {
                    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);
    }
    复制代码

在创建ClientWorker的时候构造方法中做了如下操作：

1.  初始化了本地变量和参数

1.  创建了一个远程通信的客户端

1.  配置了一个定时线程池，并且对需要认证的服务端做了认证

1.  执行配置的监听，如果没有监听器的变动，则等待5s处理一次

继续看executeConfigListen()方法。

publicvoidexecuteConfigListen() {
        // 存放含有listen的cacheData
        Map<String, List<CacheData>> listenCachesMap = newHashMap<>(16);
        // 存放不含有listen的cacheData
        Map<String, List<CacheData>> removeListenCachesMap = newHashMap<>(16);
        longnow= System.currentTimeMillis();
        // 当前时间减去上次全量同步时间是否大于5分钟booleanneedAllSync= now - lastAllSyncTime >= ALL_SYNC_INTERNAL;
        for (CacheData cache : cacheMap.get().values()) {
            synchronized (cache) {
                // 是否和服务端一致if (cache.isSyncWithServer()) {
                    // 一致则检查md5值，若md5值和上一个不一样，则说明变动了，需要通知监听器
                    cache.checkListenerMd5();
                    // 是否到全量同步时间了，未到则直接跳过if (!needAllSync) {
                        continue;
                    }
                }
                if (!cache.isDiscard()) {
                    // 非丢弃型，即新增，放入listenCachesMapif (!cache.isUseLocalConfigInfo()) {
                        List<CacheData> cacheDatas = listenCachesMap.get(String.valueOf(cache.getTaskId()));
                        if (cacheDatas == null) {
                            cacheDatas = newLinkedList<>();
                            listenCachesMap.put(String.valueOf(cache.getTaskId()), cacheDatas);
                        }
                        cacheDatas.add(cache);
                    }
                } elseif (cache.isDiscard()) {
                     // 丢弃型，即删除， 放入removeListenCachesMapif (!cache.isUseLocalConfigInfo()) {
                        List<CacheData> cacheDatas = removeListenCachesMap.get(String.valueOf(cache.getTaskId()));
                        if (cacheDatas == null) {
                            cacheDatas = newLinkedList<>();
                            removeListenCachesMap.put(String.valueOf(cache.getTaskId()), cacheDatas);
                        }
                        cacheDatas.add(cache);
                    }
                }
            }
        }
        // 此时，如果需要和服务端数据同步，则listenCachesMap和removeListenCachesMap存放了本地数据，需要和服务端对比booleanhasChangedKeys=false;
        // 新增的处理if (!listenCachesMap.isEmpty()) {
            for (Map.Entry<String, List<CacheData>> entry : listenCachesMap.entrySet()) {
                StringtaskId= entry.getKey();
                Map<String, Long> timestampMap = newHashMap<>(listenCachesMap.size() * 2);
                List<CacheData> listenCaches = entry.getValue();
                for (CacheData cacheData : listenCaches) {
                    timestampMap.put(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant),
                                     cacheData.getLastModifiedTs().longValue());
                }
                // 构建新增数据的请求参数，此请求用于远程和本地对比，发现变动了会进行通知ConfigBatchListenRequestconfigChangeListenRequest= buildConfigRequest(listenCaches);
                // 配置需要新增或更新监听数据
                configChangeListenRequest.setListen(true);
                try {
                    // 获取一个rpc的客户端RpcClientrpcClient= ensureRpcClient(taskId);
                    ConfigChangeBatchListenResponseconfigChangeBatchListenResponse= (ConfigChangeBatchListenResponse) requestProxy(
                        rpcClient, configChangeListenRequest);
                    if (configChangeBatchListenResponse.isSuccess()) {
                        Set<String> changeKeys = newHashSet<>();
                        //handle changed keys,notify listenerif (!CollectionUtils.isEmpty(configChangeBatchListenResponse.getChangedConfigs())) {
                            // 有变动的数据
                            hasChangedKeys = true;
                            for (ConfigChangeBatchListenResponse.ConfigContext changeConfig : configChangeBatchListenResponse
                                 .getChangedConfigs()) {
                                StringchangeKey= GroupKey
                                    .getKeyTenant(changeConfig.getDataId(), changeConfig.getGroup(),
                                                  changeConfig.getTenant());
                                changeKeys.add(changeKey);
                                // 刷新配置并通知变动
                                refreshContentAndCheck(changeKey);
                            }
                        }
                        //handler content configsfor (CacheData cacheData : listenCaches) {
                            StringgroupKey= GroupKey
                                .getKeyTenant(cacheData.dataId, cacheData.group, cacheData.getTenant());
                            if (!changeKeys.contains(groupKey)) {
                                //sync:cache data md5 = server md5 && cache data md5 = all listeners md5.synchronized (cacheData) {
                                    if (!cacheData.getListeners().isEmpty()) {
                                        LongpreviousTimesStamp= timestampMap.get(groupKey);
                                        if (previousTimesStamp != null && !cacheData.getLastModifiedTs()
                                            .compareAndSet(previousTimesStamp, System.currentTimeMillis())) {
                                            continue;
                                        }
                                        // 缓存数据没有变动，设置为和服务器同步
                                        cacheData.setSyncWithServer(true);
                                    }
                                }
                            }
                            // 设置为需要初始化
                            cacheData.setInitializing(false);
                        }
                    }
                } catch (Exception e) {
                    LOGGER.error("Async listen config change error ", e);
                    try {
                        Thread.sleep(50L);
                    } catch (InterruptedException interruptedException) {
                        //ignore
                    }
                }
            }
        }
        // 需要删除的数据不为空if (!removeListenCachesMap.isEmpty()) {
            for (Map.Entry<String, List<CacheData>> entry : removeListenCachesMap.entrySet()) {
                StringtaskId= entry.getKey();
                List<CacheData> removeListenCaches = entry.getValue();
                ConfigBatchListenRequestconfigChangeListenRequest= buildConfigRequest(removeListenCaches);
                // 配置需要删除
                configChangeListenRequest.setListen(false);
                try {
                    // 获取rpc客户端RpcClientrpcClient= ensureRpcClient(taskId);
                    // 通知服务端移除数据booleanremoveSuccess= unListenConfigChange(rpcClient, configChangeListenRequest);
                    if (removeSuccess) {
                        for (CacheData cacheData : removeListenCaches) {
                            synchronized (cacheData) {
                                if (cacheData.isDiscard()) {
                                    // 移除缓存
                                    ClientWorker.this
                                        .removeCache(cacheData.dataId, cacheData.group, cacheData.tenant);
                                }
                            }
                        }
                    }
                } catch (Exception e) {
                    LOGGER.error("async remove listen config change error ", e);
                }
                try {
                    Thread.sleep(50L);
                } catch (InterruptedException interruptedException) {
                    //ignore
                }
            }
        }
    
        if (needAllSync) {
            // 更新同步时间
            lastAllSyncTime = now;
        }
        //If has changed keys,notify re sync md5.if (hasChangedKeys) {
            // 服务端告知了有数据变动，则需要再同步一次
            notifyListenConfig();
        }
    }
    复制代码

上述一大段代码主要是描述的针对配置的变动的处理。触发点有两个

1.  本地的数据的变动，比如添加或者移除监听

1.  定时轮询5s检查一次本地配置(本地在3种情况下有变化，分别是第一次新增监听、接收了新的配置和监听被移除)，如果本地一直没变化，5min去全量同步一次服务端中的配置信息。

其主要是将本地的配置文件一个个去和服务端对比，发现变动了获取服务端最新的配置信息，并通知响应的监听器。这里有个几个重点方法，分别介绍下

第一个就是cache.checkListenerMd5()

// cacheData的校验方法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) {
        // 获取到监听器finalListenerlistener= listenerWrap.listener;
        if (listenerWrap.inNotifying) {
            LOGGER.warn(
                "[{}] [notify-currentSkip] dataId={}, group={}, md5={}, listener={}, listener is not finish yet,will try next time.",
                name, dataId, group, md5, listener);
            return;
        }
        // 定义一个通知任务Runnablejob= () -> {
            longstart= System.currentTimeMillis();
            ClassLoadermyClassLoader= Thread.currentThread().getContextClassLoader();
            ClassLoaderappClassLoader= listener.getClass().getClassLoader();
            try {
                if (listener instanceof AbstractSharedListener) {
                    // 扩展点，像spring cloud alibaba就用到，创建了NacosContextRefresherAbstractSharedListeneradapter= (AbstractSharedListener) listener;
                    adapter.fillContext(dataId, group);
                    LOGGER.info("[{}] [notify-context] dataId={}, group={}, md5={}", name, dataId, group, md5);
                }
                // Before executing the callback, set the thread classloader to the classloader of// the specific webapp to avoid exceptions or misuses when calling the spi interface in// the callback method (this problem occurs only in multi-application deployment).
                Thread.currentThread().setContextClassLoader(appClassLoader);
    
                ConfigResponsecr=newConfigResponse();
                cr.setDataId(dataId);
                cr.setGroup(group);
                cr.setContent(content);
                cr.setEncryptedDataKey(encryptedDataKey);
                configFilterChainManager.doFilter(null, cr);
                StringcontentTmp= cr.getContent();
                listenerWrap.inNotifying = true;
                // 回调通知，也就是通知变动的内容
                listener.receiveConfigInfo(contentTmp);
                // compare lastContent and contentif (listener instanceof AbstractConfigChangeListener) {
                    // 扩展点，告知配置内容的变动
                    Map<String, ConfigChangeItem> data = ConfigChangeHandler.getInstance()
                        .parseChangeData(listenerWrap.lastContent, contentTmp, type);
                    ConfigChangeEventevent=newConfigChangeEvent(data);
                    ((AbstractConfigChangeListener) listener).receiveConfigChange(event);
                    listenerWrap.lastContent = contentTmp;
                }
                // 赋值最新的md5
                listenerWrap.lastCallMd5 = md5;
            } catch (NacosException ex) {
                // 省略部分代码...
            } catch (Throwable t) {
                // 省略部分代码...
            } finally {
                listenerWrap.inNotifying = false;
                Thread.currentThread().setContextClassLoader(myClassLoader);
            }
        };
    
        finallongstartNotify= System.currentTimeMillis();
        try {
            if (null != listener.getExecutor()) {
                // 设置了自己监听器的异步线程池，就用自己的执行回调job
                listener.getExecutor().execute(job);
            } else {
                try {
                    // 没有设置，用内置的线程池回调
                    INTERNAL_NOTIFIER.submit(job);
                } catch (RejectedExecutionException rejectedExecutionException) {
                     // 省略部分代码...
                    job.run();
                } catch (Throwable throwable) {
                    // 省略部分代码...
                    job.run();
                }
            }
        } catch (Throwable t) {
            // 省略部分代码...
        }
        finallongfinishNotify= System.currentTimeMillis();
         // 省略部分代码...
    }
    复制代码

这个方法是比对内容的md5值，发现不一致了就会进行回调通知。当然其也做了一个扩展点的预留。俗称钩子方法。可以继承后抽象类后，自己对通知变动的一个扩展。然后它还采用了异步通知的方式。尽可能的不阻塞流程。因为回调的处理存在不确定性，为了不阻塞整个通知回调流程，采用了异步的方法。

第二个是RpcClient rpcClient = ensureRpcClient(taskId)

// 获取一个rpcClientprivate RpcClient ensureRpcClient(String taskId)throws NacosException {
        synchronized (ClientWorker.this) {
            Map<String, String> labels = getLabels();
            Map<String, String> newLabels = newHashMap<>(labels);
            newLabels.put("taskId", taskId);
            // 通过工厂创建一个客户端，这里创建了个GrpcSdkClientRpcClientrpcClient= RpcClientFactory
                .createClient(uuid + "_config-" + taskId, getConnectionType(), newLabels);
            if (rpcClient.isWaitInitiated()) {
                // 初始化网络请求处理器
                initRpcClientHandler(rpcClient);
                rpcClient.setTenant(getTenant());
                rpcClient.clientAbilities(initAbilities());
                // 启动客户端
                rpcClient.start();
            }
            return rpcClient;
        }
    }
    复制代码

先分析下initRpcClientHandler(rpcClient)

privatevoidinitRpcClientHandler(final RpcClient rpcClientInner) {
        // 注册了服务端调用客户端的处理方法，注意不是客户端请求，而是客户端接受服务端的请求，因为Grpc是可以双向请求的
        rpcClientInner.registerServerRequestHandler((request) -> {
            if (request instanceof ConfigChangeNotifyRequest) {
                // 服务端推送的请求ConfigChangeNotifyRequestconfigChangeNotifyRequest= (ConfigChangeNotifyRequest) request;
                LOGGER.info("[{}] [server-push] config changed. dataId={}, group={},tenant={}",
                            rpcClientInner.getName(), configChangeNotifyRequest.getDataId(),
                            configChangeNotifyRequest.getGroup(), configChangeNotifyRequest.getTenant());
                StringgroupKey= GroupKey
                    .getKeyTenant(configChangeNotifyRequest.getDataId(), configChangeNotifyRequest.getGroup(),
                                  configChangeNotifyRequest.getTenant());
                CacheDatacacheData= cacheMap.get().get(groupKey);
                if (cacheData != null) {
                    synchronized (cacheData) {
                        // 推送了数据后，设置最后的更新事件，并且表明和服务端不同步，通知处理
                        cacheData.getLastModifiedTs().set(System.currentTimeMillis());
                        cacheData.setSyncWithServer(false);
                        notifyListenConfig();
                    }
                }
                returnnewConfigChangeNotifyResponse();
            }
            returnnull;
        });
    
        rpcClientInner.registerServerRequestHandler((request) -> {
            if (request instanceof ClientConfigMetricRequest) {
                // 指标通知处理ClientConfigMetricResponseresponse=newClientConfigMetricResponse();
                response.setMetrics(getMetrics(((ClientConfigMetricRequest) request).getMetricsKeys()));
                return response;
            }
            returnnull;
        });
        // 处理连接和断开事件，因为能感应到断开了，所以不需要心跳检测
        rpcClientInner.registerConnectionListener(newConnectionEventListener() {
            @OverridepublicvoidonConnected() {
                LOGGER.info("[{}] Connected,notify listen context...", rpcClientInner.getName());
                notifyListenConfig();
            }
    
            @OverridepublicvoidonDisConnect() {
                StringtaskId= rpcClientInner.getLabels().get("taskId");
                LOGGER.info("[{}] DisConnected,clear listen context...", rpcClientInner.getName());
                Collection<CacheData> values = cacheMap.get().values();
                for (CacheData cacheData : values) {
                    if (StringUtils.isNotBlank(taskId)) {
                        if (Integer.valueOf(taskId).equals(cacheData.getTaskId())) {
                            cacheData.setSyncWithServer(false);
                        }
                    } else {
                        cacheData.setSyncWithServer(false);
                    }
                }
            }
    
        });
        rpcClientInner.serverListFactory(newServerListFactory() {
            @Overridepublic String genNextServer() {
                return ConfigRpcTransportClient.super.serverListManager.getNextServerAddr();
            }
            @Overridepublic String getCurrentServer() {
                return ConfigRpcTransportClient.super.serverListManager.getCurrentServerAddr();
            }
            @Overridepublic List<String> getServerList() {
                return ConfigRpcTransportClient.super.serverListManager.getServerUrls();
            }
        });
        // 增加订阅服务列表变动事件
        NotifyCenter.registerSubscriber(newSubscriber<ServerlistChangeEvent>() {
            @OverridepublicvoidonEvent(ServerlistChangeEvent event) {
                rpcClientInner.onServerListChange();
            }
            @Overridepublic Class<? extendsEvent> subscribeType() {
                return ServerlistChangeEvent.class;
            }
        });
    }
    复制代码

在初始化网络处理器的时候需要知道的是Grpc的双向处理，客户端注册了服务端请求到客户端，客户端的处理方法。并且可以对连接和断开事件做出响应。

再分析下rpcClient.start()

publicfinalvoidstart()throws NacosException {
        // 设置状态为启动状态booleansuccess= rpcClientStatus.compareAndSet(RpcClientStatus.INITIALIZED, RpcClientStatus.STARTING);
        if (!success) {
            return;
        }
        // 创建线程池
        clientEventExecutor = newScheduledThreadPoolExecutor(2, r -> {
            Threadt=newThread(r);
            t.setName("com.alibaba.nacos.client.remote.worker");
            t.setDaemon(true);
            return t;
        });
        // connection event consumer.
        clientEventExecutor.submit(() -> {
            while (!clientEventExecutor.isTerminated() && !clientEventExecutor.isShutdown()) {
                ConnectionEvent take;
                try {
                    // 处理连接事件的阻塞队列
                    take = eventLinkedBlockingQueue.take();
                    if (take.isConnected()) {
                        notifyConnected();
                    } elseif (take.isDisConnected()) {
                        notifyDisConnected();
                    }
                } catch (Throwable e) {
                    // Do nothing
                }
            }
        });
        clientEventExecutor.submit(() -> {
            while (true) {
                try {
                    if (isShutdown()) {
                        break;
                    }
                    // 处理重新连接ReconnectContextreconnectContext= reconnectionSignal
                        .poll(rpcClientConfig.connectionKeepAlive(), TimeUnit.MILLISECONDS);
                    // 省略部分代码...
                    reconnect(reconnectContext.serverInfo, reconnectContext.onRequestFail);
                } catch (Throwable throwable) {
                    // Do nothing
                }
            }
        });
        // 省略部分代码...// 注册连接重置处理器
        registerServerRequestHandler(newConnectResetRequestHandler());
        // 注册网络请求处理器
        registerServerRequestHandler(request -> {
            if (request instanceof ClientDetectionRequest) {
                returnnewClientDetectionResponse();
            }
            returnnull;
        });
    }
    复制代码

start方法是开启了线程池，对连接进行处理，包括连接，断开，重连等事件的处理。Grpc虽然是双向通信，提升性能，但是由此也更加复杂，需要管理连接，对连接事件做处理。所以，技术并没有银弹，有的是取舍。当你选择一项技术，在享受其便利性的同时也得付出一定的代价，只是这个代价是否能够接受。能否处理好其带来的影响。

## 总结 ##

本篇介绍了两个核心类CacheData和ClientWorker的一些核心方法和说明。希望帮助大家更深一步理解获取配置信息，通知，回调的机制、流程。下篇开始，我们从服务端的角度，分析下服务端是怎么处理客户端请求。

作者：ruipost

链接：[https://juejin.cn/post/7213307113111306299][https_juejin.cn_post_7213307113111306299]

[https_juejin.cn_post_7213307113111306299]: https://juejin.cn/post/7213307113111306299