Nacos 配置中心之长轮询--客户端

news2024/11/30 4:42:38

先来看下长轮询调用的链路
在这里插入图片描述

客户端

入口
在 NacosConfigService 初始化的时候,会初始化两个组件

  • 一是网络组件,也就是http数据处理的 (起作用的是 ServerHttpAgent)
  • 二是客户端的长轮询ClientWorker
    public NacosConfigService(Properties properties) throws NacosException {
        String encodeTmp = properties.getProperty(PropertyKeyConst.ENCODE);
        if (StringUtils.isBlank(encodeTmp)) {
            encode = Constants.ENCODE;
        } else {
            encode = encodeTmp.trim();
        }
        initNamespace(properties);
        // 初始化网络通信组件
        agent = new MetricsHttpAgent(new ServerHttpAgent(properties));
        agent.start();
        // 初始化ClientWorker
        worker = new ClientWorker(agent, configFilterChainManager, properties);
    }

初始化ClientWorker在初始化的时候就会初始化两个定时调度线程池,以及启动一个定时任务,该定时任务会执行ClientWorker.checkConfigInfo()方法(10ms执行一次)

    public ClientWorker(final HttpAgent agent, final ConfigFilterChainManager configFilterChainManager, final Properties properties) {
        this.agent = agent;
        this.configFilterChainManager = configFilterChainManager;

        // Initialize the timeout parameter

        init(properties);

        // 初始化一个定时调度的线程池
        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;
            }
        });

        // 初始化一个长轮询的定时调度线程池
        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;
            }
        });

        // 设置定时任务的执行频率，并且调用 checkConfigInfo 这个方法，定时去检测设置是否发生了变化
        // 严格的说是定时检测配置数量是否超过分片数量，超过了会新建一个异步任务来处理新分片的配置
        // 首次执行延迟时间为 1 毫秒、延迟时间为 10 毫秒
        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);
    }

配置文件分片处理

    public void checkConfigInfo() {
        // 分任务
        int listenerSize = cacheMap.get().size();
        // 向上取整为批数
        // 以 3000 个配置为基数分片 每个分片都有一个异步任务
        int longingTaskCount = (int) Math.ceil(listenerSize / ParamUtil.getPerTaskConfigSize());
        if (longingTaskCount > currentLongingTaskCount) {
            for (int i = (int) currentLongingTaskCount; i < longingTaskCount; i++) {
                // 要判断任务是否在执行 这块需要好好想想。 任务列表现在是无序的。变化过程可能有问题
                // 实现长轮询 这个异步任务就是来监听配置的 每个异步任务监听3000个配置
                executorService.execute(new LongPollingRunnable(i));
            }
            currentLongingTaskCount = longingTaskCount;
        }
    }

配置文件处理

   class LongPollingRunnable implements Runnable {
        private int taskId;

        public LongPollingRunnable(int taskId) {
            this.taskId = taskId;
        }

        @Override
        public void run() {

            List<CacheData> cacheDatas = new ArrayList<CacheData>();
            List<String> inInitializingCacheList = new ArrayList<String>();
            try {
                // check failover config
                // 校验本地配置同时获取同分片设置
                for (CacheData cacheData : cacheMap.get().values()) {
                    // 同分片的配置才处理
                    if (cacheData.getTaskId() == taskId) {
                        // 将同分片的配置加入集合
                        cacheDatas.add(cacheData);
                        try {
                            // 通过本地配置文件和cacheData集合中的数据进行比对，判断是否出现数据变化
                            checkLocalConfig(cacheData);
                            // 这里表示数据有变化，需要通知监听器
                            if (cacheData.isUseLocalConfigInfo()) {
                                // 通知所有针对当前配置设置了监听的监听器
                                cacheData.checkListenerMd5();
                            }
                        } catch (Exception e) {
                            LOGGER.error("get local config info error", e);
                        }
                    }
                }

                // check server config
                // 与服务端对比，找到需要更新的配置 key
                List<String> changedGroupKeys = checkUpdateDataIds(cacheDatas, inInitializingCacheList);
                LOGGER.info("get changedGroupKeys:" + changedGroupKeys);

                // 遍历发生了变化的key， 并根据key去服务端请求最新配置，并更新到内存缓存中
                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 {
                        // 从远程服务端获取最新的配置，并缓存到内存中
                        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))) {
                        // 通知所有针对当前配置设置了监听的监听器
                        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);
            }
        }
    }

向服务端对比更新的过程就类似于:我先拿着这个分片所有配置文件的key和内容的MD5去和服务端配置对比,服务端对比这些文件的MD5之后返回需要更新的配置文件key,然后客户端遍历这些需要更新的配置文件key去主动请求服务端获取最新的数据,然后更新本地缓存和本地缓存的文件.

本地配置文件与缓存数据对比
checkLocalConfig


    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);
            String md5 = MD5.getInstance().getMD5String(content);
            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);
            String md5 = MD5.getInstance().getMD5String(content);
            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));
        }
    }

上面的步骤就是:
1、如果缓存配置不使用本地配置,但是本地配置存在,则设置该缓存配置为使用本地配置(本地配置文件的内容写到缓存配置中 —> 更新 ) 需要发布变更通知
2、如果缓存配置为使用本地配置,但是本地配置不存在,则设置缓存配置为不使用本地配置,不需要通知
3、如果缓存配置为使用本地配置,本地配置存在,但是缓存配置和本地配置的版本不一样,则需要将本地配置文件写到缓存配置中更新

开启长轮询进行服务对比
checkUpdateDataIds

   List<String> checkUpdateDataIds(List<CacheData> cacheDatas, List<String> inInitializingCacheList) throws IOException {
        StringBuilder sb = new StringBuilder();
        for (CacheData cacheData : cacheDatas) {
            // 找到分片下所有不使用本地配置的缓存配置拼接成String
            if (!cacheData.isUseLocalConfigInfo()) {
                sb.append(cacheData.dataId).append(WORD_SEPARATOR);
                sb.append(cacheData.group).append(WORD_SEPARATOR);
                if (StringUtils.isBlank(cacheData.tenant)) {
                    sb.append(cacheData.getMd5()).append(LINE_SEPARATOR);
                } else {
                    sb.append(cacheData.getMd5()).append(WORD_SEPARATOR);
                    sb.append(cacheData.getTenant()).append(LINE_SEPARATOR);
                }
                if (cacheData.isInitializing()) {
                    // cacheData 首次出现在cacheMap中&首次check更新
                    inInitializingCacheList
                        .add(GroupKey.getKeyTenant(cacheData.dataId, cacheData.group, cacheData.tenant));
                }
            }
        }
        boolean isInitializingCacheList = !inInitializingCacheList.isEmpty();
        // 把这些需要配置string 发到服务端做对比
        return checkUpdateConfigStr(sb.toString(), isInitializingCacheList);
    }

ClientWorker.checkUpdateConfigStr
开启长轮询,处理对比逻辑, 长轮询请求URL:v1/ns/configs/listener

   List<String> checkUpdateConfigStr(String probeUpdateString, boolean isInitializingCacheList) throws IOException {


        List<String> params = new ArrayList<String>(2);
        params.add(Constants.PROBE_MODIFY_REQUEST);
        params.add(probeUpdateString);

        List<String> headers = new ArrayList<String>(2);
        // 长轮询标识
        headers.add("Long-Pulling-Timeout");
        headers.add("" + timeout);

        // told server do not hang me up if new initializing cacheData added in
        if (isInitializingCacheList) {
            headers.add("Long-Pulling-Timeout-No-Hangup");
            headers.add("true");
        }

        // 需要更新的配置为空就直接返回
        if (StringUtils.isBlank(probeUpdateString)) {
            return Collections.emptyList();
        }

        try {
            // In order to prevent the server from handling the delay of the client's long task,
            // increase the client's read timeout to avoid this problem
            // 超时时间 默认 30s + 15s = 45s
            long readTimeoutMs = timeout + (long) Math.round(timeout >> 1);
            // 发起长轮询
            //   请求路径： http://ip:port/nacos/v1/ns/configs/listener
            HttpResult result = agent.httpPost(Constants.CONFIG_CONTROLLER_PATH + "/listener", headers, params,
                agent.getEncode(), readTimeoutMs);

            if (HttpURLConnection.HTTP_OK == result.code) {
                setHealthServer(true);
                return parseUpdateDataIdResponse(result.content);
            } else {
                setHealthServer(false);
                LOGGER.error("[{}] [check-update] get changed dataId error, code: {}", agent.getName(), result.code);
            }
        } catch (IOException e) {
            setHealthServer(false);
            LOGGER.error("[" + agent.getName() + "] [check-update] get changed dataId exception", e);
            throw e;
        }
        return Collections.emptyList();
    }

通知监听器
checkListenerMd5

    void checkListenerMd5() {
        for (ManagerListenerWrap wrap : listeners) {
            // 对比listener 和 cacheData的MD5，如果不一样则代表cacheData发生了变化
            if (!md5.equals(wrap.lastCallMd5)) {
                // 然后触发监听器的回调处理
                safeNotifyListener(dataId, group, content, type, md5, wrap);
            }
        }
    }

md5是在什么时候变化的呢?在更新content内容变的时候,就变了
在这里插入图片描述
监听回调处理
safeNotifyListener
在该方法内，cacheData会遍历所有listener，只要两者MD5不同就会在该方法内触发调用listener.receiveConfigInfo方法
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/84944.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

本地连接docker mysql

本地连接docker mysql

1.拉取镜像 docker pull mysql 2.启动mysql实例容器 docker run --name mysql -p 3307:3306 -e MYSQL_ROOT_PASSWORDmysql_pw -d mysql --name 为mysql的实例设置别名。 -p 3307为对外暴露的端口。3306是内部端口 -e MYSQL_ROOT_PASSWORD 设置mysql登录密码 -d 以守…
阅读更多...
1.41.5 模型评估和选择，正则化和交叉验证

1.41.5 模型评估和选择，正则化和交叉验证

1.4&1.5 模型评估和选择&#xff0c;正则化和交叉验证模型评估和选择训练误差和测试误差过拟合正则化与交叉验证正则化交叉验证模型评估和选择 训练误差和测试误差 将预测系统的X作为输入&#xff0c;输入到模型里面&#xff0c;就可以得到预测结果。 学习到的模型&…
阅读更多...
UNIX网络编程卷一 学习笔记 第五章 TCP客户/服务器程序示例

UNIX网络编程卷一 学习笔记 第五章 TCP客户/服务器程序示例

本章将编写一个完整的TCP客户/服务器程序&#xff0c;这个简单例子是执行以下步骤的一个回射服务器&#xff1a; 1.客户从标准输入读入一行文本&#xff0c;并写给服务器&#xff1b; 2.服务器从网络输入读入这行文本&#xff0c;并回射给客户&#xff1b; 3.客户从网络输入读入…
阅读更多...
Spring Ioc 依赖来源-7

Spring Ioc 依赖来源-7

1. 依赖查找的来源&#xff1a;除容器内建和自定义Spring Bean之外&#xff0c;还有其他来源提供依赖查找吗&#xff1f; 查找来源 Spring 內建 BeanDefintion Spring 內建单例对象 当spring在注解环境下面, 这个 registerAnnotationConfigProcessors API会被调用, 它会被…
阅读更多...
Xilinx Vivado的RTL分析（RTL analysis）、综合（synthesis）和实现（implementation）的区别？

Xilinx Vivado的RTL分析（RTL analysis）、综合（synthesis）和实现（implementation）的区别？

1、一般流程 Xilinx 的开发工具Vivado其实还是比较好上手的&#xff0c;在左边的设计流程导航已经把FPGA的开发过程按先后顺序给排列出来了&#xff1a; Project Manager&#xff1a;项目管理器&#xff0c;此项是对项目的参数进行设置 IP Integrator&#xff1a;IP集成器&…
阅读更多...
广域铭岛参编《数智化供应链参考架构》标准正式发布

广域铭岛参编《数智化供应链参考架构》标准正式发布

近日&#xff0c;广域铭岛参编的《数智化供应链参考架构》标准正式发布。该标准由工业互联网产业联盟&#xff08;以下简称“联盟/AII”&#xff09;发布&#xff0c;是国内首个数智化供应链领域的参考架构标准&#xff0c;明确了新兴的数字化和智能化技术如何在供应链领域应用…
阅读更多...
大数据进程管理

大数据进程管理

进程管理 查看进程 进程查看命令 ps la | head -5&#xff0c;能够观察所有系统的数据 ps axjf | head -20&#xff0c;连同部分程序树状态 ps l仅查看自己的bash相关的进程 ps aux观察系统所有进程 属性含义USER进程使用者PID进程标识符%CPU进程使用掉的CPU 资源百分比%MEM…
阅读更多...
一个不错的docker支持音频的rdp桌面

一个不错的docker支持音频的rdp桌面

docker pull danielguerra/xfce4-rdp-audio 获取该镜像后&#xff0c;运行 docker run -d --name xfce4_0 --shm-size 1g -p 3389:3389 danielguerra/xfce4-rdp-audio相当于开了3个不同的系统容器&#xff0c;端口分别的13389,23389,33389&#xff0c;这时用远程桌面就可以连…
阅读更多...
记录--记一次前端CSS升级

记录--记一次前端CSS升级

这里给大家分享我在网上总结出来的一些知识&#xff0c;希望对大家有所帮助 目前平台前端使用的是原生CSSBEM命名&#xff0c;在多人协作的模式下&#xff0c;容易出现样式冲突。为了减少这一类的问题&#xff0c;提升研效&#xff0c;我调研了业界上主流的7种CSS解决方案&…
阅读更多...
Java基于springboot+vue 的传统乐器培训管理系统 elementUI

Java基于springboot+vue 的传统乐器培训管理系统 elementUI

此网站系统的开发方式和信息管理方式&#xff0c;借鉴前人设计的信息和研发。以在线乐器培训管理为主&#xff0c;以乐器培训管理为核心功能来进行设计和研发&#xff0c;把网站信息和技术整合&#xff0c;开发出一套网上乐器培训管理系统。主要运用现在社会公司中最新的技术框…
阅读更多...
rocketmq源码学习-nameServer

rocketmq源码学习-nameServer

前言 最近看了下rocketmq的源码&#xff0c;计划针对最近的学习&#xff0c;做一个笔记&#xff0c;先从nameServer启动的逻辑开始记录吧 在rocketmq中&#xff0c;有四个关键的组件 nameServerbrokerproducerconsumer 这四个组件之间的关系是这样的 关于nameSrv namese…
阅读更多...
[附源码]Python计算机毕业设计钓鱼爱好者交流平台Django（程序+LW）

[附源码]Python计算机毕业设计钓鱼爱好者交流平台Django（程序+LW）

该项目含有源码、文档、程序、数据库、配套开发软件、软件安装教程 项目运行 环境配置&#xff1a; Pychram社区版 python3.7.7 Mysql5.7 HBuilderXlist pipNavicat11Djangonodejs。 项目技术&#xff1a; django python Vue 等等组成&#xff0c;B/S模式 pychram管理等…
阅读更多...
老照片修复清晰？父母以前的老照片还能修复吗？

老照片修复清晰？父母以前的老照片还能修复吗？

父母结婚时拍摄的结婚照片&#xff0c;现在大概快四十年了&#xff0c;因为保存不善&#xff0c;导致照片泛黄&#xff0c;严重模糊。因为这是父母年轻的时候唯一保留下来的&#xff0c;对我们来说意义重大&#xff0c;所以想要修复照片可以实现吗&#xff1f; 有些照相馆是提…
阅读更多...
论文投稿指南——中国（中文EI）期刊推荐（第6期）

论文投稿指南——中国（中文EI）期刊推荐（第6期）

&#x1f680; EI是国际知名三大检索系统之一&#xff0c;在学术界的知名度和认可度仅次于SCI&#xff01;&#x1f384;&#x1f388; 【前言】 想发论文怎么办&#xff1f;手把手教你论文如何投稿&#xff01;那么&#xff0c;首先要搞懂投稿目标——论文期刊。其中&#xf…
阅读更多...
ADI Blackfin DSP处理器-BF533的开发详解51：Bin_Conver (图像二值变换处理)（含源码）

ADI Blackfin DSP处理器-BF533的开发详解51：Bin_Conver (图像二值变换处理)（含源码）

硬件准备 ADSP-EDU-BF533&#xff1a;BF533开发板 AD-HP530ICE&#xff1a;ADI DSP仿真器 软件准备 Visual DSP软件 硬件链接 功能介绍 代码实现了图像二值变换处理&#xff0c;代码运行时&#xff0c;会通过文件系统打开工程文件根目下" …/ImageView"路径中的…
阅读更多...
阿里云DataWorks荣获DAMA中国数据治理优秀产品奖

阿里云DataWorks荣获DAMA中国数据治理优秀产品奖

DAMA&#xff08;国际数据管理协会&#xff09;是一个全球性的专业组织&#xff0c;协会自1980年成立以来&#xff0c;一直致力于数据管理和数字化的研究、实践及相关知识体系的建设&#xff0c;先后发行了《DAMA 数据管理字典》和《DAMA数据管理的知识体系》等&#xff0c;该知…
阅读更多...
C/C++程序的断点调试 - Visual Studio Code

C/C++程序的断点调试 - Visual Studio Code

本文以Visual Studio Code为例&#xff0c;简述C/C程序断点调试的基本方法和过程。其它的IDE环境&#xff0c;大同小异。 本文引用自作者编写的下述图书; 本文允许以个人学习、教学等目的引用、讲授或转载&#xff0c;但需要注明原作者"海洋饼干叔 叔"&#xff1b;本…
阅读更多...
视频特效如何制作？快把这些方法收好

视频特效如何制作？快把这些方法收好

小伙伴们平时刷短视频的时候&#xff0c;有没有发现一些短视频的效果很惊艳。这些惊艳的效果&#xff0c;大部分都是在视频中添加的一些动画特效。那你们知道手机视频怎么添加特效吗&#xff1f;为了帮助大家解决这个问题&#xff0c;接下来我就将为大家分享几种添加特效的方法…
阅读更多...
架构高可用之限流-抽刀断水水更流

架构高可用之限流-抽刀断水水更流

上图中是一个水坝泄洪的图&#xff0c;那么&#xff0c;对于软件系统&#xff0c;如何使用最方便的可编程的方式增加服务限流能力呢&#xff1f; 下面我结合一个常规的springCloud项目实践了一把&#xff0c;希望他山之石可以攻玉。 背景 简单使用jmeter&#xff0c;压20个并…
阅读更多...
FL Studio21.0.0完整版最高版本升级功能有哪些?

FL Studio21.0.0完整版最高版本升级功能有哪些?

支持苹果 Silicon 芯片 – 对苹果 Silicon 芯片&#xff08;M1 芯片以及相关 CPU&#xff09;的原生 ARM 代码支持&#xff0c;但请注意&#xff1a; NewTime、NewTone 和一些 DirectWave 采样格式的导入功能尚未完全重构可能会有问题。 FL Studio-win21中文更新下载如下: htt…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023