Nacos源码 (4) 配置中心

266 0 0

作者:用户不存在！

本文阅读nacos-2.0.2的config源码，编写示例，分析推送配置、监听配置的原理。

客户端

创建NacosConfigService对象

Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.SERVER_ADDR, NACOS_HOST);
NacosConfigService configService = new NacosConfigService(properties);

构造方法：

public NacosConfigService(Properties properties) throws NacosException {
    ValidatorUtils.checkInitParam(properties);
    
    initNamespace(properties);
    this.configFilterChainManager = new ConfigFilterChainManager(properties);
    ServerListManager serverListManager = new ServerListManager(properties);
    serverListManager.start();
    
    this.worker = new ClientWorker(this.configFilterChainManager, serverListManager, properties);
    // will be deleted in 2.0 later versions
    agent = new ServerHttpAgent(serverListManager);
}

创建ConfigFilterChainManager - 过滤器链
创建ServerListManager - 服务器列表管理
创建ClientWorker - 用来发送请求，内部封装了一个ConfigRpcTransportClient类型对象agent，它能够获取到RpcClient与服务端进行通信

推送配置

Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.SERVER_ADDR, NACOS_HOST);

NacosConfigService configService = new NacosConfigService(properties);

StringWriter out = new StringWriter();
properties.store(out, "test config");

// 推送配置到nacos服务器
configService.publishConfig(
    ORDER_SERVICE, Constants.DEFAULT_GROUP, out.toString(), "properties");

推送配置到nacos服务器：

public boolean publishConfig(String dataId,
                             String group,
                             String content,
                             String type) throws NacosException {
    return publishConfigInner(namespace, dataId, group, null, null, null, content, type, null);
}

private boolean publishConfigInner(String tenant, String dataId, String group, String tag, String appName,
        String betaIps, String content, String type, String casMd5) throws NacosException {
    group = blank2defaultGroup(group);
    ParamUtils.checkParam(dataId, group, content);

    ConfigRequest cr = new ConfigRequest();
    cr.setDataId(dataId);
    cr.setTenant(tenant);
    cr.setGroup(group);
    cr.setContent(content);
    cr.setType(type);
    configFilterChainManager.doFilter(cr, null);
    content = cr.getContent();
    String encryptedDataKey = (String) cr.getParameter("encryptedDataKey");

    return worker.publishConfig(
        dataId, group, tenant, appName, tag, betaIps, content, encryptedDataKey, casMd5, type);
}

worker使用agent推送配置：

// 1. 封装ConfigPublishRequest对象
ConfigPublishRequest request = new ConfigPublishRequest(dataId, group, tenant, content);
request.setCasMd5(casMd5);
request.putAdditionalParam(TAG_PARAM, tag);
request.putAdditionalParam(APP_NAME_PARAM, appName);
request.putAdditionalParam(BETAIPS_PARAM, betaIps);
request.putAdditionalParam(TYPE_PARAM, type);
request.putAdditionalParam(ENCRYPTED_DATA_KEY_PARAM, encryptedDataKey);
// 2. 获取RpcClient对象
// 3. 使用RpcClient发请求
ConfigPublishResponse response = (ConfigPublishResponse) requestProxy(getOneRunningClient(), request);

监听配置

Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.SERVER_ADDR, NACOS_HOST);

NacosConfigService configService = new NacosConfigService(properties);

// 添加监听器
configService.addListener(ORDER_SERVICE, Constants.DEFAULT_GROUP, new AbstractListener() {
  @Override
  public void receiveConfigInfo(String configInfo) {
    System.out.printf(">> config: n%snn", configInfo);
  }
});

将监听器注册到本地，本地使用CacheData作为监听器管理器，封装配置名、dataId和监听器集合，内部使用CopyOnWriteArrayList保存监听器，如果是第一次监听，会先拉取一次配置。本地注册表为Map结构，使用dataId+group+tenant作为key，value是CacheData对象
在client初始化阶段，会注册一个ServerRequestHandler，专门处理服务器端的ConfigChangeNotifyRequest请求，该请求只会推送变化了的配置的基本信息，而不包括内容，所以此处还会触发一次ConfigBatchListenRequest请求
之后查找到变化的配置后，再发一个查询请求拉取配置

服务端

推送配置处理器

ConfigPublishRequestHandler处理器

配置中心使用ConfigPublishRequestHandler类处理客户端配置推送请求：

验证请求参数
封装configAdvanceInfo配置扩展信息
创建ConfigInfo封装配置信息
使用持久层对象insert或者update配置

使用ConfigChangePublisher推送一个ConfigDataChangeEvent事件

ConfigChangePublisher.notifyConfigChange(
    new ConfigDataChangeEvent(false, dataId, group, tenant, time.getTime()));

ConfigDataChangeEvent事件处理

ConfigDataChangeEvent事件会触发数据dump操作和集群同步操作。

此处先介绍数据dump操作：

dumpService.dump(syncRequest.getDataId(), syncRequest.getGroup(), syncRequest.getTenant(),
    syncRequest.getTag(), syncRequest.getLastModified(), NetUtils.localIP());

dumpService是Dump data service用于备份数据：

// Add DumpTask to TaskManager, it will execute asynchronously.
public void dump(String dataId, String group, String tenant, String tag, long lastModified, String handleIp,
        boolean isBeta) {
    String groupKey = GroupKey2.getKey(dataId, group, tenant);
    String taskKey = String.join("+", dataId, group, tenant, String.valueOf(isBeta), tag);
    // 推送一个DumpTask
    dumpTaskMgr.addTask(taskKey, new DumpTask(groupKey, tag, lastModified, handleIp, isBeta));
}

DumpTask将使用DumpProcessor类处理：

把数据写到磁盘
推送LocalDataChangeEvent事件

LocalDataChangeEvent事件会被以下几个类处理：

LongPollingService$1会触发DataChangeTask，响应长轮询订阅者
RpcConfigChangeNotifier将数据变化推送给rpc订阅者
InternalConfigChangeNotifier处理内部配置文件变化

监听配置处理器

ConfigChangeBatchListenRequestHandler处理器

处理客户端的配置监听请求：

public class ConfigChangeBatchListenRequestHandler
        extends RequestHandler<ConfigBatchListenRequest, ConfigChangeBatchListenResponse> {

    @Autowired
    private ConfigChangeListenContext configChangeListenContext;

    @TpsControl(pointName = "ConfigListen")
    @Secured(action = ActionTypes.READ, parser = ConfigResourceParser.class)
    public ConfigChangeBatchListenResponse handle(
            ConfigBatchListenRequest configChangeListenRequest, RequestMeta meta)
            throws NacosException {
        String connectionId = StringPool.get(meta.getConnectionId());
        String tag = configChangeListenRequest.getHeader(Constants.VIPSERVER_TAG);

        ConfigChangeBatchListenResponse configChangeBatchListenResponse = 
            new ConfigChangeBatchListenResponse();
        for (ConfigBatchListenRequest.ConfigListenContext listenContext : configChangeListenRequest
                .getConfigListenContexts()) {
            String groupKey = GroupKey2
                    .getKey(listenContext.getDataId(), listenContext.getGroup(), listenContext.getTenant());
            groupKey = StringPool.get(groupKey);

            String md5 = StringPool.get(listenContext.getMd5());

            // 监听
            if (configChangeListenRequest.isListen()) {
                // 注册监听器，维护groupKey->connectionId集关系和groupKey->md5关系
                configChangeListenContext.addListen(groupKey, md5, connectionId);
                // 判断变化
                boolean isUptoDate = ConfigCacheService.isUptodate(groupKey, md5, meta.getClientIp(), tag);
                if (!isUptoDate) {
                    // 把变化的配置基本信息添加到响应
                    configChangeBatchListenResponse
                        .addChangeConfig(listenContext.getDataId(), listenContext.getGroup(),
                            listenContext.getTenant());
                }
            } else {
                // 取消监听
                configChangeListenContext.removeListen(groupKey, connectionId);
            }
        }

        return configChangeBatchListenResponse;
    }
}

内容来源于网络如有侵权请私信删除

文章来源: 博客园

原文链接: https://www.cnblogs.com/xugf/p/17647070.html

标签： java java8 java开发

你还没有登录，请先登录或注册！

还没有人评论，欢迎说说您的想法！

Nacos源码 (4) 配置中心

Nacos源码 (4) 配置中心

客户端

创建NacosConfigService对象

推送配置

监听配置

服务端

推送配置处理器

ConfigPublishRequestHandler处理器

ConfigDataChangeEvent事件处理

监听配置处理器

ConfigChangeBatchListenRequestHandler处理器

相关课程

热门标签

Nacos源码 (4) 配置中心

Nacos源码 (4) 配置中心

客户端

创建NacosConfigService对象

推送配置

监听配置

服务端

推送配置处理器

ConfigPublishRequestHandler处理器

ConfigDataChangeEvent事件处理

监听配置处理器

ConfigChangeBatchListenRequestHandler处理器

相关课程

热门标签

推荐文章