Nacos源码分析- 01 ConfigService创建流程

Posted 2021-04-05 Reactor2020

基本API

/**
 * 测试从 Nacos Server 拉取配置内容
 * @throws NacosException
 */
@Test
public void getConfigTest() throws Exception {

    //1、配置server-addr
    Properties properties = new Properties();
    properties.put(PropertyKeyConst.SERVER_ADDR, "127.0.0.1:8848,192.168.1.1:9999");

    //2、创建ConfigService对象
    ConfigService configService = NacosFactory.createConfigService(properties);

 //3、通过ConfigService对Nacos Server上的配置进行操作，比如拉取配置、修改配置等
    String content = configService.getConfig("other", "DEFAULT_GROUP", 5000L);
    System.out.println("content:" + content);
    configService.publishConfig("other", "DEFAULT_GROUP", "===修改后内容===");
}

上面是使用编程方式对Nacos Server上的配置进行操作，从上述代码可以看出，客户端对Nacos Server上的配置文件进行的操作都被抽象到ConfigService接口中，该接口定义如下：

public interface ConfigService {

    //获取配置配置
    String getConfig(String dataId, String group, long timeoutMs) throws NacosException;

    //获取配置同时添加Listener，用于监听变更事件
    String getConfigAndSignListener(String dataId, String group, long timeoutMs, Listener listener) throws NacosException;

    //添加Listener，用于监听变更事件
    void addListener(String dataId, String group, Listener listener) throws NacosException;

    //发布配置
    boolean publishConfig(String dataId, String group, String content) throws NacosException;

    //移除配置
    boolean removeConfig(String dataId, String group) throws NacosException;

    //移除Listener
    void removeListener(String dataId, String group, Listener listener);

    //获取Server状态 UP or DOWN
    String getServerStatus();
    
}

NacosConfigService

ConfigService是客户端对Nacos Server配置操作顶层抽象接口，并提供唯一实现类NacosConfigService，通过分析NacosConfigService源码可以理解Nacos作为配置中心和客户端交互机制，这一节我们首先来分析下NacosConfigService创建时到底处理了哪些事。

NacosFactory.createConfigService(properties)采用工厂设计模式，内部通过反射方式创建NacosConfigService对象。

public NacosConfigService(Properties properties) throws NacosException {
    //初始化encode
    String encodeTmp = properties.getProperty(PropertyKeyConst.ENCODE);
    if (StringUtils.isBlank(encodeTmp)) {
        encode = Constants.ENCODE;
    } else {
        encode = encodeTmp.trim();
    }
    //初始化namespace，并放置到properties中
    initNamespace(properties);

    //ServerHttpAgent使用http方式访问nacos server，并使用装饰模式，MetricsHttpAgent具有指标统计功能
    agent = new MetricsHttpAgent(new ServerHttpAgent(properties));
    //ServerHttpAgent.start()会调用ServerListManager.start()方法，完成nacos ip list收集
    agent.start();

    worker = new ClientWorker(agent, configFilterChainManager, properties);
}

大致流程：

• 初始化 encode，主要用于指定 http请求中编码使用的字符集；
• 初始化 namespace，主要处理兼容阿里云中的多租户概念；
• 创建 ServerHttpAgent对象，作为 Nacos Server在客户端的 agent，封装向 Nacos发送 http请求的相关底层操作，然后采用装饰模式包装成 MetricsHttpAgent具有指标统计功能，实现对发送的 http请求进行统计；
• agent.start()：最终调用 ServerListManager.start()方法， ServerListManager主要用于管理 nacos ip列表， nacos ip列表可以采用配置方式指定，还可以定时从某个 url地址获取， ServerListManager.start()主要就是启动定时任务周期性的从某个 url接口中获取 nacos ip列表；
• new ClientWorker(agent, configFilterChainManager, properties)：

ServerHttpAgent

public ServerHttpAgent(Properties properties) throws NacosException {
    // ServerListManager主要用于管理server urls
    serverListMgr = new ServerListManager(properties);
    // SecurityProxy用于管理登录
    securityProxy = new SecurityProxy(properties);
    // 获取namespace
    namespaceId = properties.getProperty(PropertyKeyConst.NAMESPACE);
    // 初始化encode、ak/sk、maxRetry
    init(properties);
    /**
     * 如果配置username，这里会调用/v1/auth/users/login进行登录，服务端返回accessToken
     * 后续ServerHttpAgent向Nacos发送请求时都会把accessToken放置到header中带过去
     */
    securityProxy.login(serverListMgr.getServerUrls());

    ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);
            t.setName("com.alibaba.nacos.client.config.security.updater");
            t.setDaemon(true);
            return t;
        }
    });

    /**
     * 定时任务5秒执行一次login()，符合一定条件时向nacos发送login请求获取新的accessToken，避免token失效
     */
    executorService.scheduleWithFixedDelay(new Runnable() {
        @Override
        public void run() {
            securityProxy.login(serverListMgr.getServerUrls());
        }
    }, 0, securityInfoRefreshIntervalMills, TimeUnit.MILLISECONDS);
}

大致流程：

• 创建 ServerListManager对象，用于管理 nacos ip列表，主要分为两类：固定方式和动态方式，固定方式根据配置解析 nacos ip列表，而动态方式会指定 url，当调用 ServerListManager.start()时开启定时任务周期从 url请求 nacos ip列表。
• 创建 SecurityProxy，当配置 username时向 nacos发送 login请求，获取 accessToken，后续 ServerHttpAgent向 nacos发送 http请求时都会把 accessToken放置到 header中带过去。
• accessToken具有时效性，开启一个定时任务周期向 nacos发送 login请求获取新 accessToken，避免 token失效。

客户端和nacos server之间交互使用http协议，HttpAgent接口定义了这些交互方式，其核心定义如下：

public interface HttpAgent {
    //获取nacos ip列表，主要是动态方式下启动定时任务周期获取最新数据，
    //固定方式创建时就解析完成，这里一般没有操作
    void start() throws NacosException;

    //http get方式向nacos发送请求
    HttpResult httpGet(String path, List<String> headers, List<String> paramValues, String encoding, long readTimeoutMs) throws IOException;

    //http post方式向nacos发送请求
    HttpResult httpPost(String path, List<String> headers, List<String> paramValues, String encoding, long readTimeoutMs) throws IOException;

    //http delete方式向nacos发送请求
    HttpResult httpDelete(String path, List<String> headers, List<String> paramValues, String encoding, long readTimeoutMs) throws IOException;

}

HttpAgent的实现类ServerHttpAgent，start方法委托给了ServerListManager，这里主要看下httpGet()、httpPost()、httpDelete()这几个方法，其逻辑是一样的，大致逻辑：

1. 调用 serverListMgr.getCurrentServerAddr()获取一个 nacos ip，并赋值给 currentServerAddr，表示当前发送请求的服务地址；
2. 然后对 currentServerAddr代表的服务发送 http请求，如果成功则 currentServerAddr不变，然后返回请求结果；
3. 如果请求失败，则通过 serverListMgr.getIterator().next()获取下一个 nacos服务地址，然后发送 http请求；
4. 如果 serverListMgr维护的列表都执行一遍后依然没有成功，则 maxRetry减1后小于0则抛出 ConnectException异常，否则调用 serverListMgr.refreshCurrentServerAddr()打乱 serverUrls之后再重新遍历一次，重复上面步骤2和步骤3;
5. 避免长时间重试，可以通过 readTimeoutMs设置超时时间，如果超时则抛出异常 ConnectException("no available server")；
6. 如果以上有请求成功，则把请求成功的地址改成 currentServerAddr。

ClientWorker

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

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

大致流程：

• this.agent = agent：底层向 nacos发送 http请求使用的是 HttpAgent对象；
• configFilterChainManager： ConfigFilter链管理器，获取到配置信息后触发 ConfigFilter过滤器用于扩展；
• 创建线程池，每10毫秒执行一次 checkConfigInfo()，可以通过 addListener(String dataId, String group, Listener listener)添加配置变更监听，当检测到配置变更时会触发 Listener回掉通知， checkConfigInfo()方法中就是处理监听配置变更逻辑代码。

总结

前面分析了NacosConfigService、ClientWorker、ServerHttpAgent、ServerListManager等几个关键类，它们之间关系如下图：

• Nacos客户端和 Nacos服务端之间使用 http协议进行交互，这些交互实现被封装到 ServerHttpAgent中，典型方法是： httpGet()、httpPost()、httpDelete()，主要实现了请求失败时重试机制。
• ServerHttpAgent内部持有一个 ServerListManager对象，主要用于管理 Nacos ip列表， Nacos ip信息可以通过配置添加，还可以通过 url地址定时动态获取。 ServerHttpAgent负责向 Nacos发送 http请求，具体发送给哪个 Nacos Server，以及发送失败如何选择其它 Nacos Server等都是由 ServerListManager负责。
• ServerHttpAgent只负责向 Nacos发送 http请求， ClientWorker则是基于它进行了相关业务操作封装，比如获取配置文件、监听配置变更通知等，配置相关操作大部分基本都是由 ClientWorker完成。
• NacosConfigService则是作为配置中心时提供给开发者使用的一个门面类，开发者通过 NacosConfigService可以实现对配置的各种操作。

             长按识别关注，持续输出原创