源码分析Nacos如何动态刷新配置_Java

一 nacos 刷新配置的源码阅读

在 clientworker 中配置了定义了一个的内部类 longpollingrunnable 并实现了runnable 接口直接到 cachedata.checklistenermd5() 这个方法

public void run() {
             // 获取定义的group
            list<cachedata> cachedatas = new arraylist<cachedata>();
            list<string> ininitializingcachelist = new arraylist<string>();
            try {
                // check failover config
                for (cachedata cachedata : cachemap.values()) {
                    if (cachedata.gettaskid() == taskid) {
                        cachedatas.add(cachedata);
                        try {
                            checklocalconfig(cachedata);
                            if (cachedata.isuselocalconfiginfo()) {
                                cachedata.checklistenermd5();
                            }
                        } catch (exception e) {
                            logger.error("get local config info error", e);
                        }
                    }
                }
            
                // check server config
                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 {
                        string[] ct = getserverconfig(dataid, group, tenant, 3000l);
                        cachedata cache = cachemap.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);
            }
        }
    }

检查当前的md5值是否更改

void checklistenermd5() {
        for (managerlistenerwrap wrap : listeners) {
            if (!md5.equals(wrap.lastcallmd5)) {
              // 如果md5值变了，就发送对应事件通知
                safenotifylistener(dataid, group, content, type, md5, wrap);
            }
        }
    }

安全的通知监听器配置改变：

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);
                }

                // 设置线程上下文类加载器为应用加载器（避免多应用部署时，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();

                // 调用监听器的 receiveconfiginfo 方法通知变更
                listener.receiveconfiginfo(contenttmp);

                // 如果是支持配置变更事件的监听器，触发对应事件
                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;
                }

                // 更新上一次调用的 md5 值
                listenerwrap.lastcallmd5 = md5;

                // 打印通知成功日志
                logger.info("[{}] [notify-ok] dataid={}, group={}, md5={}, listener={} ", name, dataid, group, md5,
                        listener);
            } catch (nacosexception ex) {
                // 特定 nacos 异常处理
                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);
}

nacoscontextrefresher 中 registernacoslistenersforapplications的方法

/**
 * 为指定的 dataid + group 注册一个 nacos 配置监听器
 * @param groupkey 配置分组（group）
 * @param datakey 配置标识（dataid）
 */
private void registernacoslistener(final string groupkey, final string datakey) {
    // 生成一个唯一 key，用于标识监听器（key = group + "++" + dataid）
    string key = nacospropertysourcerepository.getmapkey(datakey, groupkey);

    // 从 listenermap 中获取对应 key 的监听器，如果不存在则创建一个 abstractsharedlistener
    listener listener = listenermap.computeifabsent(key,
        lst -> new abstractsharedlistener() {
            /**
             * 当配置变更时，会触发该方法
             */
            @override
            public void innerreceive(string dataid, string group, string configinfo) {
                // 刷新次数 +1（用于监控/统计）
                refreshcountincrement();

                // 记录刷新历史
                nacosrefreshhistory.addrefreshrecord(dataid, group, configinfo);

                // 发布 spring 的 refreshevent，通知上下文环境配置已变更
                // 注意：这里是全量刷新，
                applicationcontext.publishevent(
                    new refreshevent(this, null, "refresh nacos config"));

                // 如果开启了 debug 日志，打印变更信息
                if (log.isdebugenabled()) {
                    log.debug(string.format(
                        "refresh nacos config group=%s,dataid=%s,configinfo=%s",
                        group, dataid, configinfo));
                }
            }
        });

    try {
        // 调用 nacos 客户端 api，注册监听器
        configservice.addlistener(datakey, groupkey, listener);
    }
    catch (nacosexception e) {
        // 注册失败，记录警告日志
        log.warn(string.format(
            "register fail for nacos listener ,dataid=[%s],group=[%s]", datakey,
            groupkey), e);
    }
}

在springclould 中的 refresheventlistener

public void onapplicationevent(applicationevent event) {
		if (event instanceof applicationreadyevent) {
			handle((applicationreadyevent) event);
		}
		else if (event instanceof refreshevent) {
			handle((refreshevent) event);
		}
	}

public void handle(refreshevent event) {
	if (this.ready.get()) { // don't handle events before app is ready
		log.debug("event received " + event.geteventdesc());
		set<string> keys = this.refresh.refresh();
		log.info("refresh keys changed: " + keys);
	}
}

contextrefresher 中的 refresh 方法刷新所有作用域为 refresh 的bean

public synchronized set<string> refresh() {
		set<string> keys = refreshenvironment();
        // 刷新所有的
		this.scope.refreshall();
		return keys;
	}

二 @refreshscope注解

定义

@refreshscope 是 spring cloud 提供的注解，主要用于 支持配置的动态刷新，特别是在结合像 nacos、consul、spring cloud config 等配置中心时使用。@refreshscope 使得标注的 bean 在配置变更并发布刷新事件时，能够被重新实例化，从而实现“配置热更新”。

使用背景

spring boot 默认的 bean 是单例的（@singleton），一旦初始化完成，其属性就不会再变化。如果你想在运行时通过配置中心动态刷新某个 bean 中的属性，就必须加上 @refreshscope

与nacos配合使用demo

1、依赖引入

确保你引入了以下依赖（以 spring boot 2.x / spring cloud alibaba 为例）：

<dependency>
    <groupid>com.alibaba.cloud</groupid>
    <artifactid>spring-cloud-starter-alibaba-nacos-config</artifactid>
</dependency>

2、application.yml配置

server:
  port: 8080

spring:
  application:
    name: nacos-refresh-demo
  cloud:
    nacos:
      config:
        server-addr: 127.0.0.1:8848
        file-extension: yaml
        group: default_group
        namespace: public
        refresh-enabled: true

3、编写配置类（使用 @refreshscope）

package com.example.nacosdemo.config;

import lombok.data;
import org.springframework.boot.context.properties.configurationproperties;
import org.springframework.cloud.context.config.annotation.refreshscope;
import org.springframework.stereotype.component;

@data
@component
@refreshscope // 开启动态刷新
@configurationproperties(prefix = "custom")
public class customconfig {
    private string name;
    private integer age;
}

4、测试 controller

package com.example.nacosdemo.controller;

import com.example.nacosdemo.config.customconfig;
import org.springframework.web.bind.annotation.getmapping;
import org.springframework.web.bind.annotation.restcontroller;

@restcontroller
@requiredargsconstructor
public class testcontroller {

    private final customconfig customconfig;

    @getmapping("/config")
    public string getconfig() {
        return "name: " + customconfig.getname() + ", age: " + customconfig.getage();
    }
}

然后更改你的nacos中的配置，查看是否被更新呢