nacos配置中心之服务器端

配置信息的发布
- 配置信息发布请求URL: POST: /v1/cs/configs
- nacos在STANDALONE模式或集群模式没有指定用mysql情况下使用derby数据库，在集群模式且指定mysql情况下使用mysql
- 持续化到数据库后立马同步集群中的其他节点

    /**

     * 增加或更新非聚合数据。

     *

     * @throws NacosException

     */

    @PostMapping

    @Secured(action = ActionTypes.WRITE, parser = ConfigResourceParser.class)

    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 {

        //省略参数验证代码

        .....................

        //persistService.insertOrUpdate插入或更新数据库

        //EventDispatcher.fireEvent 通知集群中其他节点

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

        }

        return true;

    }

EventDispatcher.fireEvent 触发 AsyncNotifyService.onEvent事件
AsyncNotifyService.onEvent事件中遍历serverList通知集群各个节点
如果某个节点不健康或同步失败延迟一段时间后重新通知：schedule(asyncTask, delay, TimeUnit.MILLISECONDS);



@Service

public class AsyncNotifyService extends AbstractEventListener {

    //onEvent事件遍历serverList通知集群各个节点

    @Override

    public void onEvent(Event event) {

        // 并发产生 ConfigDataChangeEvent

        if (event instanceof ConfigDataChangeEvent) {

            ConfigDataChangeEvent evt = (ConfigDataChangeEvent) event;

            long dumpTs = evt.lastModifiedTs;

            String dataId = evt.dataId;

            String group = evt.group;

            String tenant = evt.tenant;

            String tag = evt.tag;

            List<?> ipList = serverListService.getServerList();

            // 其实这里任何类型队列都可以

            Queue<NotifySingleTask> queue = new LinkedList<NotifySingleTask>();

            for (int i = 0; i < ipList.size(); i++) {

                queue.add(new NotifySingleTask(dataId, group, tenant, tag, dumpTs, (String) ipList.get(i), evt.isBeta));

            }

            EXECUTOR.execute(new AsyncTask(httpclient, queue));

        }

    }

   // 同步任务逻辑

   class AsyncTask implements Runnable {

        @Override

        public void run() {

            executeAsyncInvoke();

        }

        private void executeAsyncInvoke() {

            while (!queue.isEmpty()) {

                NotifySingleTask task = queue.poll();

                String targetIp = task.getTargetIP();

                if (serverListService.getServerList().contains(

                    targetIp)) {

                    // 启动健康检查且有不监控的ip则直接把放到通知队列，否则通知

                    if (serverListService.isHealthCheck() && ServerListService.getServerListUnhealth().contains(targetIp)) {

                        // target ip 不健康，则放入通知列表中

                        // get delay time and set fail count to the task

                        // schedule(asyncTask, delay, TimeUnit.MILLISECONDS);

                        asyncTaskExecute(task);

                    } else {

                        // /v1/cs/communication/dataChange?dataId={2}&group={3}

                        HttpGet request = new HttpGet(task.url);

                        request.setHeader(NotifyService.NOTIFY_HEADER_LAST_MODIFIED, String.valueOf(task.getLastModified()));

                        request.setHeader(NotifyService.NOTIFY_HEADER_OP_HANDLE_IP, LOCAL_IP);

                        httpclient.execute(request, new AsyncNotifyCallBack(httpclient, task));

                    }

                }

            }

        }

    }

    class AsyncNotifyCallBack implements FutureCallback<HttpResponse> {

      // 同步其他节点失败

      @Override

      public void failed(Exception ex) {

            long delayed = System.currentTimeMillis() - task.getLastModified();

            //get delay time and set fail count to the task

            //延迟一段时间后重新通知

            //schedule(asyncTask, delay, TimeUnit.MILLISECONDS);

            asyncTaskExecute(task);

            MetricsMonitor.getConfigNotifyException().increment();

        }

     }

}

配置信息的获取

读取配置信息的时候加读锁
1. lockResult>0 加锁成功，获取配置信息后返回HttpServletResponse.SC_OK 或 HttpServletResponse.SC_NOT_FOUND
2. lockResult==0对应的配置信息不存在返回 HttpServletResponse.SC_NOT_FOUND
3. lockResult<0 已有线程加了写锁返回 HttpServletResponse.SC_CONFLICT
单机且不用mysql情况下从数据库中获取，其他配置从文件中获取配置信息(本地文件相当于mysql的缓存）



@Service

public class ConfigServletInner {

   /**

     * 同步配置获取接口

     */

    public String doGetConfig(HttpServletRequest request, HttpServletResponse response, String dataId, String group,

                              String tenant, String tag, String clientIp) throws IOException, ServletException {

        int lockResult = tryConfigReadLock(groupKey);

        if (lockResult > 0) {

           //代码挺多总体逻辑归纳如下

           //单机且不用mysql情况下从数据库中获取，其他配置从文件中获取配置信息

           //数据库情况下判断从config_info,还是config_info_beta，还是config_info_tag表获取数据

           //文件情况下判断从还是BASE_DIR还是BETA_DIR,还是TAG_DIR目录获取配置文件

           //本地文件获取比mysql中获取更快，本地文件相当于mysql的缓存

        }

    }

}

为了查询效率，mysql中的数据缓存到每个集群节点的文件中，缓存时机有3中情况

当nacos启动的时候如果当前时间距离最后一次心跳时间超过6个小时，则全量缓存mysql数据
启动定时任务每6个小时重新全量缓存mysql数据
@GetMapping("/dataChange")时处理单个数据

 //@PostConstruct init() 调用 dumpConfigInfo

 private void dumpConfigInfo(DumpAllProcessor dumpAllProcessor) throws IOException {

        int timeStep = 6;

        Boolean isAllDump = true;

        // initial dump all

        FileInputStream fis = null;

        Timestamp heartheatLastStamp = null;

        try {

            if (isQuickStart()) {

                File heartbeatFile = DiskUtil.heartBeatFile();

                if (heartbeatFile.exists()) {

                    fis = new FileInputStream(heartbeatFile);

                    String heartheatTempLast = IoUtils.toString(fis, Constants.ENCODE);

                    heartheatLastStamp = Timestamp.valueOf(heartheatTempLast);

                    //当前时间距离最后一次心跳时间超过6个小时

                    if (TimeUtils.getCurrentTime().getTime() - heartheatLastStamp.getTime() < timeStep * 60 * 60 * 1000) {

                        isAllDump = false;

                    }

                }

            }

            if (isAllDump) {

                LogUtil.defaultLog.info("start clear all config-info.");

                DiskUtil.clearAll();

                //

                dumpAllProcessor.process(DumpAllTask.TASK_ID, new DumpAllTask());

            }

        } catch (IOException e) {

            LogUtil.fatalLog.error("dump config fail" + e.getMessage());

            throw e;

        }

    }

 @PostConstruct

 public void init() {

       //启动定时任务定时缓存数据

       TimerTaskService.scheduleWithFixedDelay(dumpAll, initialDelay, DUMP_ALL_INTERVAL_IN_MINUTE, TimeUnit.MINUTES);

 }

   /**

     * 通知配置信息改变

     */

    @GetMapping("/dataChange")

    public Boolean notifyConfigInfo(HttpServletRequest request,

                                    @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) {

        dataId = dataId.trim();

        group = group.trim();

        String lastModified = request.getHeader(NotifyService.NOTIFY_HEADER_LAST_MODIFIED);

        long lastModifiedTs = StringUtils.isEmpty(lastModified) ? -1 : Long.parseLong(lastModified);

        String handleIp = request.getHeader(NotifyService.NOTIFY_HEADER_OP_HANDLE_IP);

        String isBetaStr = request.getHeader("isBeta");

        //其他节点通知变更后缓存数据

        if (StringUtils.isNotBlank(isBetaStr) && trueStr.equals(isBetaStr)) {

            dumpService.dump(dataId, group, tenant, lastModifiedTs, handleIp, true);

        } else {

            dumpService.dump(dataId, group, tenant, tag, lastModifiedTs, handleIp);

        }

        return true;

    }

配置信息的轮询

巴特西

nacos配置中心之服务器端

最新文章

热门文章