配置中心 -- Nacos 客户端配置获取及监听实现原理

╰半夏微凉° 2021-12-15 10:31 1205阅读 0赞

这篇博客我们通过示例来了解学习一下 Nacos 的客户端获取配置以及服务端发送配置的相关流程。

### 服务端创建配置 ###

在 Nacos 的管理页面我们可以创建如下配置信息

1、在默认命名空间下新建配置

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly90aWFuanVud2VpLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70][]

2、编辑配置并发布

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly90aWFuanVud2VpLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 1][]

### 客户端获取配置 ###

**1、客户端主动获取配置**

（1）客户端创建 ConfigService 服务根据 dataId 和 group 来获取配置信息

**示例：根据dataId 和 group 获取相关配置**

（2）nacos  服务端在 ConfigController 中提供接口 getConfig 来提供相关配置查询接口

@RequestMapping(method = RequestMethod.GET)
        public void getConfig(HttpServletRequest request, HttpServletResponse response,
                              @RequestParam("dataId") String dataId, @RequestParam("group") String group,
                              @RequestParam(value = "tenant", required = false, defaultValue = StringUtils.EMPTY)
                                  String tenant,
                              @RequestParam(value = "tag", required = false) String tag)
            throws IOException, ServletException, NacosException {
            // check params
            ParamUtils.checkParam(dataId, group, "datumId", "content");
            ParamUtils.checkParam(tag);
    
            final String clientIp = RequestUtil.getRemoteIp(request);
            inner.doGetConfig(request, response, dataId, group, tenant, tag, clientIp);
        }

在 ConfigServletInner 的 doGetConfig 方法中会根据 dataId、group、tenant、tag 等进行查询

public String doGetConfig(HttpServletRequest request, HttpServletResponse response, String dataId, String group,
                                  String tenant, String tag, String clientIp) throws IOException, ServletException {
            //省略部分代码
            ConfigInfoBase configInfoBase = null;
    		//省略部分代码
    		configInfoBase = persistService.findConfigInfo4Beta(dataId, group, tenant);
    		//省略部分代码
                    
        }

在 PersistService 的findConfigInfo4Beta 方法中会执行 SQL 从库中查找相关配置信息

/**
         * 根据dataId和group查询配置信息
         */
        public ConfigInfo4Beta findConfigInfo4Beta(final String dataId, final String group, final String tenant) {
            String tenantTmp = StringUtils.isBlank(tenant) ? StringUtils.EMPTY : tenant;
            try {
                return this.jt.queryForObject(
                    "SELECT ID,data_id,group_id,tenant_id,app_name,content,beta_ips FROM config_info_beta WHERE data_id=?"
                        + " AND group_id=? AND tenant_id=?",
                    new Object[] {dataId, group, tenantTmp}, CONFIG_INFO4BETA_ROW_MAPPER);
            } catch (EmptyResultDataAccessException e) { // 表明数据不存在, 返回null
                return null;
            } catch (CannotGetJdbcConnectionException e) {
                fatalLog.error("[db-error] " + e.toString(), e);
                throw e;
            }
        }

**2、客户端监听配置**

客户端在启动时候从注册中心获取配置外，也可以主动监听配置中心相关的配置变化

示例代码：

package com.alibaba.nacos.example;
    
    import java.util.Properties;
    import java.util.concurrent.Executor;
    import com.alibaba.nacos.api.NacosFactory;
    import com.alibaba.nacos.api.config.ConfigService;
    import com.alibaba.nacos.api.config.listener.Listener;
    import com.alibaba.nacos.api.exception.NacosException;
    
    /**
     * Config service example
     * 
     * @author Nacos
     *
     */
    public class ConfigExample {
    
    	public static void main(String[] args) throws NacosException, InterruptedException {
    		String serverAddr = "localhost";
    		String dataId = "mysql";
    		String group = "DEFAULT_GROUP";
    		Properties properties = new Properties();
    		properties.put(PropertyKeyConst.SERVER_ADDR, serverAddr);
    		ConfigService configService = NacosFactory.createConfigService(properties);
    		String content = configService.getConfig(dataId, group, 5000);
    		System.out.println(content);
              //注册监听
    		configService.addListener(dataId, group, new Listener() {
    			@Override
    			public void receiveConfigInfo(String configInfo) {
    				System.out.println("recieve:" + configInfo);
    			}
    
    			@Override
    			public Executor getExecutor() {
    				return null;
    			}
    		});
    		
    		boolean isPublishOk = configService.publishConfig(dataId, group, "content");
    		System.out.println(isPublishOk);
    		
    		Thread.sleep(3000);
    		content = configService.getConfig(dataId, group, 5000);
    		System.out.println(content);
    
    		boolean isRemoveOk = configService.removeConfig(dataId, group);
    		System.out.println(isRemoveOk);
    		Thread.sleep(3000);
    
    		content = configService.getConfig(dataId, group, 5000);
    		System.out.println(content);
    		Thread.sleep(300000);
    
    	}
    }

1、客户端调用 ConfigController 的 listener 的监听接口，与服务端通过 http 建立长连接来监听配置变更

configService.addListener(dataId, group, new Listener() {
                @Override
                public void receiveConfigInfo(String configInfo) {
                    System.out.println("recieve:" + configInfo);
                }
    
                @Override
                public Executor getExecutor() {
                    return null;
                }
            });

2、Nacos  服务提供接口 listener 用于客户端注册监听服务（还是基于web 容器的长连接技术）

/**
         * 比较MD5
         */
        @RequestMapping(value = "/listener", method = RequestMethod.POST)
        public void listener(HttpServletRequest request, HttpServletResponse response)
            throws ServletException, IOException {
            request.setAttribute("org.apache.catalina.ASYNC_SUPPORTED", true);
            String probeModify = request.getParameter("Listening-Configs");
            if (StringUtils.isBlank(probeModify)) {
                throw new IllegalArgumentException("invalid probeModify");
            }
    
            probeModify = URLDecoder.decode(probeModify, Constants.ENCODE);
    
            Map<String, String> clientMd5Map;
            try {
                clientMd5Map = MD5Util.getClientMd5Map(probeModify);
            } catch (Throwable e) {
                throw new IllegalArgumentException("invalid probeModify");
            }
    
            // do long-polling
            inner.doPollingConfig(request, response, clientMd5Map, probeModify.length());
        }

在 ConfigServletInner 的 doPollingConfig 会根据客户端请求获取 AsyncContext，然后异步建立长连接

/**
         * 轮询接口
         */
        public String doPollingConfig(HttpServletRequest request, HttpServletResponse response,
                                      Map<String, String> clientMd5Map, int probeRequestSize)
            throws IOException, ServletException {
    
            // 长轮询
            if (LongPollingService.isSupportLongPolling(request)) {
                longPollingService.addLongPollingClient(request, response, clientMd5Map, probeRequestSize);
                return HttpServletResponse.SC_OK + "";
            }
    
            // else 兼容短轮询逻辑
            List<String> changedGroups = MD5Util.compareMd5(request, response, clientMd5Map);
    
            // 兼容短轮询result
            String oldResult = MD5Util.compareMd5OldResult(changedGroups);
            String newResult = MD5Util.compareMd5ResultString(changedGroups);
    
            String version = request.getHeader(Constants.CLIENT_VERSION_HEADER);
            if (version == null) {
                version = "2.0.0";
            }
            int versionNum = Protocol.getVersionNumber(version);
    
            /**
             * 2.0.4版本以前, 返回值放入header中
             */
            if (versionNum < START_LONGPOLLING_VERSION_NUM) {
                response.addHeader(Constants.PROBE_MODIFY_RESPONSE, oldResult);
                response.addHeader(Constants.PROBE_MODIFY_RESPONSE_NEW, newResult);
            } else {
                request.setAttribute("content", newResult);
            }
    
            // 禁用缓存
            response.setHeader("Pragma", "no-cache");
            response.setDateHeader("Expires", 0);
            response.setHeader("Cache-Control", "no-cache,no-store");
            response.setStatus(HttpServletResponse.SC_OK);
            return HttpServletResponse.SC_OK + "";
        }

在 LongPollingService 的 addLongPollingClient 中从请求中获取 AsyncContext，然后异步建立长连接

public void addLongPollingClient(HttpServletRequest req, HttpServletResponse rsp, Map<String, String> clientMd5Map,
                                         int probeRequestSize) {
    
            String str = req.getHeader(LongPollingService.LONG_POLLING_HEADER);
            String noHangUpFlag = req.getHeader(LongPollingService.LONG_POLLING_NO_HANG_UP_HEADER);
            String appName = req.getHeader(RequestUtil.CLIENT_APPNAME_HEADER);
            String tag = req.getHeader("Vipserver-Tag");
            int delayTime = SwitchService.getSwitchInteger(SwitchService.FIXED_DELAY_TIME, 500);
            /**
             * 提前500ms返回响应，为避免客户端超时 @qiaoyi.dingqy 2013.10.22改动  add delay time for LoadBalance
             */
            long timeout = Math.max(10000, Long.parseLong(str) - delayTime);
            if (isFixedPolling()) {
                timeout = Math.max(10000, getFixedPollingInterval());
                // do nothing but set fix polling timeout
            } else {
                long start = System.currentTimeMillis();
                List<String> changedGroups = MD5Util.compareMd5(req, rsp, clientMd5Map);
                if (changedGroups.size() > 0) {
                    generateResponse(req, rsp, changedGroups);
                    LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}|{}",
                        System.currentTimeMillis() - start, "instant", RequestUtil.getRemoteIp(req), "polling",
                        clientMd5Map.size(), probeRequestSize, changedGroups.size());
                    return;
                } else if (noHangUpFlag != null && noHangUpFlag.equalsIgnoreCase(TRUE_STR)) {
                    LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}|{}", System.currentTimeMillis() - start, "nohangup",
                        RequestUtil.getRemoteIp(req), "polling", clientMd5Map.size(), probeRequestSize,
                        changedGroups.size());
                    return;
                }
            }
            String ip = RequestUtil.getRemoteIp(req);
            // 一定要由HTTP线程调用，否则离开后容器会立即发送响应
            final AsyncContext asyncContext = req.startAsync();
            // AsyncContext.setTimeout()的超时时间不准，所以只能自己控制
            asyncContext.setTimeout(0L);
    
            scheduler.execute(
                new ClientLongPolling(asyncContext, clientMd5Map, ip, probeRequestSize, timeout, appName, tag));
        }

在ClientLongPolling 中会将客户端信息添加到 长轮询订阅关系队列  中

/**
         * 长轮询订阅关系
         */
        final Queue<ClientLongPolling> allSubs;

在  ClientLongPolling 执行时添加客户端长连接到 allSubs 中

class ClientLongPolling implements Runnable {
    
            @Override
            public void run() {
                asyncTimeoutFuture = scheduler.schedule(new Runnable() {
                    public void run() {
                        try {
                            getRetainIps().put(ClientLongPolling.this.ip, System.currentTimeMillis());
                            /**
                             * 删除订阅关系
                             */
                            allSubs.remove(ClientLongPolling.this);
    
                            if (isFixedPolling()) {
                                LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}",
                                    (System.currentTimeMillis() - createTime),
                                    "fix", RequestUtil.getRemoteIp((HttpServletRequest)asyncContext.getRequest()),
                                    "polling",
                                    clientMd5Map.size(), probeRequestSize);
                                List<String> changedGroups = MD5Util.compareMd5(
                                    (HttpServletRequest)asyncContext.getRequest(),
                                    (HttpServletResponse)asyncContext.getResponse(), clientMd5Map);
                                if (changedGroups.size() > 0) {
                                    sendResponse(changedGroups);
                                } else {
                                    sendResponse(null);
                                }
                            } else {
                                LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}",
                                    (System.currentTimeMillis() - createTime),
                                    "timeout", RequestUtil.getRemoteIp((HttpServletRequest)asyncContext.getRequest()),
                                    "polling",
                                    clientMd5Map.size(), probeRequestSize);
                                sendResponse(null);
                            }
                        } catch (Throwable t) {
                            LogUtil.defaultLog.error("long polling error:" + t.getMessage(), t.getCause());
                        }
    
                    }
                }, timeoutTime, TimeUnit.MILLISECONDS);
    
                allSubs.add(this);
            }
    
            void sendResponse(List<String> changedGroups) {
                /**
                 *  取消超时任务
                 */
                if (null != asyncTimeoutFuture) {
                    asyncTimeoutFuture.cancel(false);
                }
                generateResponse(changedGroups);
            }
    
            void generateResponse(List<String> changedGroups) {
                if (null == changedGroups) {
                    /**
                     * 告诉容器发送HTTP响应
                     */
                    asyncContext.complete();
                    return;
                }
    
                HttpServletResponse response = (HttpServletResponse)asyncContext.getResponse();
    
                try {
                    String respString = MD5Util.compareMd5ResultString(changedGroups);
    
                    // 禁用缓存
                    response.setHeader("Pragma", "no-cache");
                    response.setDateHeader("Expires", 0);
                    response.setHeader("Cache-Control", "no-cache,no-store");
                    response.setStatus(HttpServletResponse.SC_OK);
                    response.getWriter().println(respString);
                    asyncContext.complete();
                } catch (Exception se) {
                    pullLog.error(se.toString(), se);
                    asyncContext.complete();
                }
            }
    
            ClientLongPolling(AsyncContext ac, Map<String, String> clientMd5Map, String ip, int probeRequestSize,
                              long timeoutTime, String appName, String tag) {
                this.asyncContext = ac;
                this.clientMd5Map = clientMd5Map;
                this.probeRequestSize = probeRequestSize;
                this.createTime = System.currentTimeMillis();
                this.ip = ip;
                this.timeoutTime = timeoutTime;
                this.appName = appName;
                this.tag = tag;
            }
    
            // =================
    
            final AsyncContext asyncContext;
            final Map<String, String> clientMd5Map;
            final long createTime;
            final String ip;
            final String appName;
            final String tag;
            final int probeRequestSize;
            final long timeoutTime;
    
            Future<?> asyncTimeoutFuture;
        }

3、配置修改发布时调用 ConfigController 的 publishConfig 接口给监听者发送消息

当用户修改配置并发布时会调用 ConfigController 的 publishConfig 方法，将配置信息入库并转发发布事件，通知建立长连接的客户端。

/**
         * 增加或更新非聚合数据。
         *
         * @throws NacosException
         */
        @RequestMapping(method = RequestMethod.POST)
        @ResponseBody
        public Boolean publishConfig(HttpServletRequest request, HttpServletResponse response,
                                     @RequestParam("dataId") String dataId, @RequestParam("group") String group,
                                     @RequestParam(value = "tenant", required = false, defaultValue = StringUtils.EMPTY)
                                         String tenant,
                                     @RequestParam("content") String content,
                                     @RequestParam(value = "tag", required = false) String tag,
                                     @RequestParam(value = "appName", required = false) String appName,
                                     @RequestParam(value = "src_user", required = false) String srcUser,
                                     @RequestParam(value = "config_tags", required = false) String configTags,
                                     @RequestParam(value = "desc", required = false) String desc,
                                     @RequestParam(value = "use", required = false) String use,
                                     @RequestParam(value = "effect", required = false) String effect,
                                     @RequestParam(value = "type", required = false) String type,
                                     @RequestParam(value = "schema", required = false) String schema)
            throws NacosException {
            final String srcIp = RequestUtil.getRemoteIp(request);
            String requestIpApp = RequestUtil.getAppName(request);
            ParamUtils.checkParam(dataId, group, "datumId", content);
            ParamUtils.checkParam(tag);
    
            Map<String, Object> configAdvanceInfo = new HashMap<String, Object>(10);
            if (configTags != null) {
                configAdvanceInfo.put("config_tags", configTags);
            }
            if (desc != null) {
                configAdvanceInfo.put("desc", desc);
            }
            if (use != null) {
                configAdvanceInfo.put("use", use);
            }
            if (effect != null) {
                configAdvanceInfo.put("effect", effect);
            }
            if (type != null) {
                configAdvanceInfo.put("type", type);
            }
            if (schema != null) {
                configAdvanceInfo.put("schema", schema);
            }
            ParamUtils.checkParam(configAdvanceInfo);
    
            if (AggrWhitelist.isAggrDataId(dataId)) {
                log.warn("[aggr-conflict] {} attemp to publish single data, {}, {}",
                    RequestUtil.getRemoteIp(request), dataId, group);
                throw new NacosException(NacosException.NO_RIGHT, "dataId:" + dataId + " is aggr");
            }
    
            final Timestamp time = TimeUtils.getCurrentTime();
            String betaIps = request.getHeader("betaIps");
            ConfigInfo configInfo = new ConfigInfo(dataId, group, tenant, appName, content);
            if (StringUtils.isBlank(betaIps)) {
                if (StringUtils.isBlank(tag)) {
                    persistService.insertOrUpdate(srcIp, srcUser, configInfo, time, configAdvanceInfo, false);
                    EventDispatcher.fireEvent(new ConfigDataChangeEvent(false, dataId, group, tenant, time.getTime()));
                } else {
                    persistService.insertOrUpdateTag(configInfo, tag, srcIp, srcUser, time, false);
                    EventDispatcher.fireEvent(new ConfigDataChangeEvent(false, dataId, group, tenant, tag, time.getTime()));
                }
            } else { // beta publish
                persistService.insertOrUpdateBeta(configInfo, betaIps, srcIp, srcUser, time, false);
                EventDispatcher.fireEvent(new ConfigDataChangeEvent(true, dataId, group, tenant, time.getTime()));
            }
            ConfigTraceService.logPersistenceEvent(dataId, group, tenant, requestIpApp, time.getTime(),
                LOCAL_IP, ConfigTraceService.PERSISTENCE_EVENT_PUB, content);
    
            return true;
        }

在 EventDispatcher的 fireEvent 方法中会根据 根据事件类型获取所有的监听器，然后调用监听器的通知方法 onEvent

static public void fireEvent(Event event) {
            if (null == event) {
                throw new IllegalArgumentException();
            }
    
            for (AbstractEventListener listener : getEntry(event.getClass()).listeners) {
                try {
                    listener.onEvent(event);
                } catch (Exception e) {
                    log.error(e.toString(), e);
                }
            }
        }

在 最终调用AbstractEventListener 的实现类LongPollingService 的 onEvent 方法

@Override
        public void onEvent(Event event) {
            if (isFixedPolling()) {
                // ignore
            } else {
                if (event instanceof LocalDataChangeEvent) {
                    LocalDataChangeEvent evt = (LocalDataChangeEvent)event;
                    scheduler.execute(new DataChangeTask(evt.groupKey, evt.isBeta, evt.betaIps));
                }
            }
        }

在 DataChangeTask 中会从 allSubs中获取所有的长连接客户端，然后通知客户端配置变更

class DataChangeTask implements Runnable {
            @Override
            public void run() {
                try {
                    ConfigService.getContentBetaMd5(groupKey);
                    for (Iterator<ClientLongPolling> iter = allSubs.iterator(); iter.hasNext(); ) {
                        ClientLongPolling clientSub = iter.next();
                        if (clientSub.clientMd5Map.containsKey(groupKey)) {
                            // 如果beta发布且不在beta列表直接跳过
                            if (isBeta && !betaIps.contains(clientSub.ip)) {
                                continue;
                            }
    
                            // 如果tag发布且不在tag列表直接跳过
                            if (StringUtils.isNotBlank(tag) && !tag.equals(clientSub.tag)) {
                                continue;
                            }
    
                            getRetainIps().put(clientSub.ip, System.currentTimeMillis());
                            iter.remove(); // 删除订阅关系
                            LogUtil.clientLog.info("{}|{}|{}|{}|{}|{}|{}",
                                (System.currentTimeMillis() - changeTime),
                                "in-advance",
                                RequestUtil.getRemoteIp((HttpServletRequest)clientSub.asyncContext.getRequest()),
                                "polling",
                                clientSub.clientMd5Map.size(), clientSub.probeRequestSize, groupKey);
                            clientSub.sendResponse(Arrays.asList(groupKey));
                        }
                    }
                } catch (Throwable t) {
                    LogUtil.defaultLog.error("data change error:" + t.getMessage(), t.getCause());
                }
            }
    
            DataChangeTask(String groupKey) {
                this(groupKey, false, null);
            }
    
            DataChangeTask(String groupKey, boolean isBeta, List<String> betaIps) {
                this(groupKey, isBeta, betaIps, null);
            }

总结：

（1）客户端可以在启动时第一次拉取配置信息

（2）客户端可以利用 web 容器的长连接机制与服务端建立长连接，服务端将订阅轮询存放在 allSubs 队列中

（3）用户修改配置发布时会从 allSubs 中取出相关长连接客户端今天配置变更通知。

（4）客户端获取到通知之后会主动请求接口获取最新的配置信息。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly90aWFuanVud2VpLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70]: /images/20211214/4b505c11b24f4f4785b2273428e0287f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly90aWFuanVud2VpLmJsb2cuY3Nkbi5uZXQ_size_16_color_FFFFFF_t_70 1]: /images/20211214/842774e620ac4f68ad3c09c577f4be11.png