• nacos上的注册过的服务实例掉线分析

    最近生产上的xxl_job框架的一个执行器（nacos客户端）因为分配内存不大，导致频繁与nacos服务端的连接断开，而断开之后虽然客户端服务没有宕掉，但是就是无法重新注册到nacos的服务端上去。

    问题定位：

    查看服务器日志；发现日志中打印好多内存溢出的情况，回顾上个迭代同事新增了一个功能，就是定时同步底层数据。后面定位问题的时候发现这个定时任务要执行12个小时才会结束，跟踪代码的时候发现每次遍历的时候要初始化大量数据到jvm内存中，这就导致了内存资源紧张，后台日志不断报内存溢出和GC回收异常的问题。

    解决方案：

    （1）针对有问题的定时业务逻辑重新进行编码优化实现。
    （2）扩大jvm分配给程序的启动内存。-Xms2g -Xmx2g统一改成 -Xms6g -Xmx6g.
    通过这2个方案整改后，1周内生产再也没有出过类似的问题。

    问题解决了，但是我想深究一下为什么nacos掉线后，就注册不上去了呢

    然后我想出现这个问题的原因是自己的服务在jvm的分配的内存使用完毕后，在后台运行的向nacos服务端保持心跳的线程阻塞或者被杀死了，导致后面nacos服务器接收不到来自客户端的心跳，从而我的服务后面没有再次被nacos服务端发现。

    基于以上情况，我试着从nacos客户端注册与心跳检测方面跟一下源码。

    首先最重要的是要找到保持服务于nacos服务端保持心跳的源码，看看这个后台运行的保持心跳的线程的运行机制。

    我猜想这个保持心跳的线程应该会在服务首次启动注册的时候激活。

    所以先找到注册的接口，去到nacos官网，找到api接口界面的注册接口：https://nacos.io/zh-cn/docs/open-api.html

    找到注册接口请求路径为/nacos/v1/ns/instance

    然后去gitlab下载nacos的源码，我直接下载的最新的，然后根据/nacos/v1/ns/instance 找到如下方法：
    可以看到服务注册成功后，还执行了个方法addBeatInfo

    1. @Override
    2.     public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
    3.  
    4.         if (instance.isEphemeral()) {
    5.             BeatInfo beatInfo = new BeatInfo();
    6.             beatInfo.setServiceName(NamingUtils.getGroupedName(serviceName, groupName));
    7.             beatInfo.setIp(instance.getIp());
    8.             beatInfo.setPort(instance.getPort());
    9.             beatInfo.setCluster(instance.getClusterName());
    10.             beatInfo.setWeight(instance.getWeight());
    11.             beatInfo.setMetadata(instance.getMetadata());
    12.             beatInfo.setScheduled(false);
    13.             beatInfo.setPeriod(instance.getInstanceHeartBeatInterval());
    14.  
    15.             beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);
    16.         }

    看一下这个方法是做什么的,可以看到BeatReactor是个线程池类

    1. public BeatReactor(NamingProxy serverProxy, int threadCount) {
    2.         this.serverProxy = serverProxy;
    3.         executorService = new ScheduledThreadPoolExecutor(threadCount, new ThreadFactory() {
    4.             @Override
    5.             public Thread newThread(Runnable r) {
    6.                 Thread thread = new Thread(r);
    7.                 thread.setDaemon(true);
    8.                 thread.setName("com.alibaba.nacos.naming.beat.sender");
    9.                 return thread;
    10.             }
    11.         });
    12.     }
    13.  
    14.     public void **addBeatInfo**(String serviceName, BeatInfo beatInfo) {
    15.         NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
    16.         String key = buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort());
    17.         BeatInfo existBeat = null;
    18.         //fix #1733
    19.         if ((existBeat = dom2Beat.remove(key)) != null) {
    20.             existBeat.setStopped(true);
    21.         }
    22.         dom2Beat.put(key, beatInfo);
    23.         //这个方法就是启动的后台保持心跳的线程
    24.         executorService.schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS);
    25.         MetricsMonitor.getDom2BeatSizeMonitor().set(dom2Beat.size());
    26.     }

    看看
    schedule(new BeatTask(beatInfo), beatInfo.getPeriod(), TimeUnit.MILLISECONDS)
    这个方法的定义

    1. /**
    2.      * Creates and executes a one-shot action that becomes enabled
    3.      * after the given delay.
    4.      *
    5.      * @param command the task to execute
    6.      * @param delay the time from now to delay execution
    7.      * @param unit the time unit of the delay parameter
    8.      * @return a ScheduledFuture representing pending completion of
    9.      *         the task and whose {@code get()} method will return
    10.      *         {@code null} upon completion
    11.      * @throws RejectedExecutionException if the task cannot be
    12.      *         scheduled for execution
    13.      * @throws NullPointerException if command is null
    14.      */
    15.     public ScheduledFuture> schedule(Runnable command,
    16.                                        long delay, TimeUnit unit);

    然后重点部分就是看BeatTask，BeatTask继承Runnable，run方法就是我们的重点，该方法调用了NamingProxy的sendBeat方法，服务端请求地址为/instance/beat的方法

    1. class BeatTask implements Runnable {
    2.  
    3.         BeatInfo beatInfo;
    4.  
    5.         public BeatTask(BeatInfo beatInfo) {
    6.             this.beatInfo = beatInfo;
    7.         }
    8.  
    9.         @Override
    10.         public void run() {
    11.             if (beatInfo.isStopped()) {
    12.                 return;
    13.             }
    14.             long nextTime = beatInfo.getPeriod();
    15.             try {
    16.                 JSONObject result = serverProxy.sendBeat(beatInfo, BeatReactor.this.lightBeatEnabled);
    17.                 long interval = result.getIntValue("clientBeatInterval");
    18.                 boolean lightBeatEnabled = false;
    19.                 if (result.containsKey(CommonParams.LIGHT_BEAT_ENABLED)) {
    20.                     lightBeatEnabled = result.getBooleanValue(CommonParams.LIGHT_BEAT_ENABLED);
    21.                 }
    22.                 BeatReactor.this.lightBeatEnabled = lightBeatEnabled;
    23.                 if (interval > 0) {
    24.                     nextTime = interval;
    25.                 }
    26.                 int code = NamingResponseCode.OK;
    27.                 if (result.containsKey(CommonParams.CODE)) {
    28.                     code = result.getIntValue(CommonParams.CODE);
    29.                 }
    30.                 if (code == NamingResponseCode.RESOURCE_NOT_FOUND) {
    31.                     Instance instance = new Instance();
    32.                     instance.setPort(beatInfo.getPort());
    33.                     instance.setIp(beatInfo.getIp());
    34.                     instance.setWeight(beatInfo.getWeight());
    35.                     instance.setMetadata(beatInfo.getMetadata());
    36.                     instance.setClusterName(beatInfo.getCluster());
    37.                     instance.setServiceName(beatInfo.getServiceName());
    38.                     instance.setInstanceId(instance.getInstanceId());
    39.                     instance.setEphemeral(true);
    40.                     try {
    41.                         serverProxy.registerService(beatInfo.getServiceName(),
    42.                             NamingUtils.getGroupName(beatInfo.getServiceName()), instance);
    43.                     } catch (Exception ignore) {
    44.                     }
    45.                 }
    46.             } catch (NacosException ne) {
    47.                 NAMING_LOGGER.error("[CLIENT-BEAT] failed to send beat: {}, code: {}, msg: {}",
    48.                     JSON.toJSONString(beatInfo), ne.getErrCode(), ne.getErrMsg());
    49.  
    50.             }
    51.             executorService.schedule(new BeatTask(beatInfo), nextTime, TimeUnit.MILLISECONDS);
    52.         }
    53.     }

    找一下这个线程是每隔几秒运行一次的，是5秒

    接下来我们把目光放到服务端，找到InstanceController的beat方法，如果是参数beat信息的话，说明是第一次发起心跳，则会带有服务实例信息，因为发起心跳成功则服务端会返回下次不要带beat信息的参数，这样客户端第二次就不会携带beat信息了。如果发现没有该服务，又没带beat信息，说明这个服务可能被移除过了，直接返回没找到。如果没有服务，但是发现有beat信息，那就从beat中获取服务实例信息，进行注册.

    1. @CanDistro
    2. @PutMapping("/beat")
    3. @Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
    4. public ObjectNode beat(HttpServletRequest request) throws Exception {
    5.  
    6.     ObjectNode result = JacksonUtils.createEmptyJsonNode();
    7.     //设置心跳间隔
    8.     result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, switchDomain.getClientBeatInterval());
    9.  
    10.     String beat = WebUtils.optional(request, "beat", StringUtils.EMPTY);
    11.     RsInfo clientBeat = null;
    12.     //判断有无心跳内容
    13.     //如果存在心跳内容则不是轻量级心跳就转化为RsInfo
    14.     if (StringUtils.isNotBlank(beat)) {
    15.         clientBeat = JacksonUtils.toObj(beat, RsInfo.class);
    16.     }
    17.     String clusterName = WebUtils
    18.             .optional(request, CommonParams.CLUSTER_NAME, UtilsAndCommons.DEFAULT_CLUSTER_NAME);
    19.     String ip = WebUtils.optional(request, "ip", StringUtils.EMPTY);
    20.     int port = Integer.parseInt(WebUtils.optional(request, "port", "0"));
    21.     if (clientBeat != null) {
    22.         if (StringUtils.isNotBlank(clientBeat.getCluster())) {
    23.             clusterName = clientBeat.getCluster();
    24.         } else {
    25.             // fix #2533
    26.             clientBeat.setCluster(clusterName);
    27.         }
    28.         ip = clientBeat.getIp();
    29.         port = clientBeat.getPort();
    30.     }
    31.     String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
    32.     String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
    33.     NamingUtils.checkServiceNameFormat(serviceName);
    34.     Loggers.SRV_LOG.debug("[CLIENT-BEAT] full arguments: beat: {}, serviceName: {}", clientBeat, serviceName);
    35.     //获取实例的信息
    36.     Instance instance = serviceManager.getInstance(namespaceId, serviceName, clusterName, ip, port);
    37.     //如果实例不存在
    38.     if (instance == null) {
    39.         if (clientBeat == null) {
    40.             result.put(CommonParams.CODE, NamingResponseCode.RESOURCE_NOT_FOUND);
    41.             return result;
    42.         }
    43.  
    44.         Loggers.SRV_LOG.warn("[CLIENT-BEAT] The instance has been removed for health mechanism, "
    45.                 + "perform data compensation operations, beat: {}, serviceName: {}", clientBeat, serviceName);
    46.         //根据您心跳内容创建一个实例信息
    47.         instance = new Instance();
    48.         instance.setPort(clientBeat.getPort());
    49.         instance.setIp(clientBeat.getIp());
    50.         instance.setWeight(clientBeat.getWeight());
    51.         instance.setMetadata(clientBeat.getMetadata());
    52.         instance.setClusterName(clusterName);
    53.         instance.setServiceName(serviceName);
    54.         instance.setInstanceId(instance.getInstanceId());
    55.         instance.setEphemeral(clientBeat.isEphemeral());
    56.         //注册实例
    57.         serviceManager.registerInstance(namespaceId, serviceName, instance);
    58.     }
    59.     //获取服务的信息
    60.     Service service = serviceManager.getService(namespaceId, serviceName);
    61.  
    62.     if (service == null) {
    63.         throw new NacosException(NacosException.SERVER_ERROR,
    64.                 "service not found: " + serviceName + "@" + namespaceId);
    65.     }
    66.     //不存在的话，要创建一个进行处理
    67.     if (clientBeat == null) {
    68.         clientBeat = new RsInfo();
    69.         clientBeat.setIp(ip);
    70.         clientBeat.setPort(port);
    71.         clientBeat.setCluster(clusterName);
    72.     }
    73.     //开启心跳检查任务
    74.     service.processClientBeat(clientBeat);
    75.  
    76.     result.put(CommonParams.CODE, NamingResponseCode.OK);
    77.     //5秒间隔
    78.     if (instance.containsMetadata(PreservedMetadataKeys.HEART_BEAT_INTERVAL)) {
    79.         result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, instance.getInstanceHeartBeatInterval());
    80.     }
    81.     //告诉客户端不需要带上心跳信息了，变成轻量级心跳了
    82.     result.put(SwitchEntry.LIGHT_BEAT_ENABLED, switchDomain.isLightBeatEnabled());
    83.     return result;
    84. }

    接下来我们看一下nacos服务端 开启心跳检查任务processClientBeat方法，该方法将ClientBeatProcessor放入到线程池中，接下来我们看下重点看下run方法

    1. public void processClientBeat(final RsInfo rsInfo) {
    2.         ClientBeatProcessor clientBeatProcessor = new ClientBeatProcessor();
    3.         clientBeatProcessor.setService(this);
    4.         clientBeatProcessor.setRsInfo(rsInfo);
    5.        //放入线程池中执行 HealthCheckReactor.scheduleNow(clientBeatProcessor);
    6.     }

    更新每次客户端发送的最新的心跳时间，这个主要是是放到内存map中

    1. @Override
    2.     public void run() {
    3.         Service service = this.service;
    4.         if (Loggers.EVT_LOG.isDebugEnabled()) {
    5.             Loggers.EVT_LOG.debug("[CLIENT-BEAT] processing beat: {}", rsInfo.toString());
    6.         }
    7.  
    8.         String ip = rsInfo.getIp();
    9.         String clusterName = rsInfo.getCluster();
    10.         int port = rsInfo.getPort();
    11.         Cluster cluster = service.getClusterMap().get(clusterName);
    12.         List<Instance> instances = cluster.allIPs(true);
    13.  
    14.         for (Instance instance : instances) {
    15.             if (instance.getIp().equals(ip) && instance.getPort() == port) {
    16.                 if (Loggers.EVT_LOG.isDebugEnabled()) {
    17.                     Loggers.EVT_LOG.debug("[CLIENT-BEAT] refresh beat: {}", rsInfo.toString());
    18.                 }
    19.        //更新每次客户端发送的最新的心跳时间，这个主要是是放到内存map中         instance.setLastBeat(System.currentTimeMillis());
    20.                 if (!instance.isMarked()) {
    21.                     if (!instance.isHealthy()) {
    22.                         instance.setHealthy(true);
    23.                         Loggers.EVT_LOG.info("service: {} {POS} {IP-ENABLED} valid: {}:{}@{}, region: {}, msg: client beat ok",
    24.                             cluster.getService().getName(), ip, port, cluster.getName(), UtilsAndCommons.LOCALHOST_SITE);
    25.                         getPushService().serviceChanged(service);
    26.                     }
    27.                 }
    28.             }
    29.         }
    30.     }

    那么，为什么nacos服务端为什么要设置注册到nacos上的服务最新的更新时间呢？

    这涉及到nacos的健康检查机制

    Nacos Server会开启一个定时任务来检查注册服务的健康情况，对于超过15秒没收到客户端的心跳实例会将它的 healthy属性置为false，此时当客户端不会将该实例的信息发现，如果某个服务的实例超过30秒没收到心跳，则剔除该实例，如果剔除的实例恢复，发送心跳则会恢复。

    当有实例注册的时候，我们会看到有个service.init()的方法，该方法的实现主要是将ClientBeatCheckTask加入到线程池当中：

    1. private void putServiceAndInit(Service service) throws NacosException {
    2.         putService(service);
    3.         /启动服务检查
    4.         service.init();
    5.         consistencyService.listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true), service);
    6.         consistencyService.listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), false), service);
    7.         Loggers.SRV_LOG.info("[NEW-SERVICE] {}", service.toJSON());
    8.     }
    9.     
    10.     
    11.  public void init() {
    12.  
    13.         HealthCheckReactor.scheduleCheck(clientBeatCheckTask);
    14.  
    15.         for (Map.Entry entry : clusterMap.entrySet()) {
    16.             entry.getValue().setService(this);
    17.             entry.getValue().init();
    18.         }
    19.     }

    ClientBeatCheckTask中的run方法主要做两件事心跳时间超过15秒则设置该实例信息为不健康状况和心跳时间超过30秒则删除该实例信息，如下代码:

    1. @Override
    2.     public void run() {
    3.         try {
    4.             if (!getDistroMapper().responsible(service.getName())) {
    5.                 return;
    6.             }
    7.  
    8.             if (!getSwitchDomain().isHealthCheckEnabled()) {
    9.                 return;
    10.             }
    11.  
    12.             List<Instance> instances = service.allIPs(true);
    13.  
    14.             // first set health status of instances:
    15.             for (Instance instance : instances) {
    16.                         //如果心跳时间超过15秒则设置该实例信息为不健康状况
    17.                 if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) {
    18.                     if (!instance.isMarked()) {
    19.                         if (instance.isHealthy()) {
    20.                             instance.setHealthy(false);
    21.                             Loggers.EVT_LOG.info("{POS} {IP-DISABLED} valid: {}:{}@{}@{}, region: {}, msg: client timeout after {}, last beat: {}",
    22.                                 instance.getIp(), instance.getPort(), instance.getClusterName(), service.getName(),
    23.                                 UtilsAndCommons.LOCALHOST_SITE, instance.getInstanceHeartBeatTimeOut(), instance.getLastBeat());
    24.                             getPushService().serviceChanged(service);
    25.                             SpringContext.getAppContext().publishEvent(new InstanceHeartbeatTimeoutEvent(this, instance));
    26.                         }
    27.                     }
    28.                 }
    29.             }
    30.  
    31.  
    32.             if (!getGlobalConfig().isExpireInstance()) {
    33.                 return;
    34.             }
    35.  
    36.             // then remove obsolete instances:
    37.             for (Instance instance : instances) {
    38.  
    39.                 if (instance.isMarked()) {
    40.                     continue;
    41.                 }
    42.             //如果心跳时间超过30秒则删除该实例信息
    43.  
    44.                 if (System.currentTimeMillis() - instance.getLastBeat() > instance.getIpDeleteTimeout()) {
    45.                     // delete instance
    46.                     Loggers.SRV_LOG.info("[AUTO-DELETE-IP] service: {}, ip: {}", service.getName(), JSON.toJSONString(instance));
    47.                     deleteIP(instance);
    48.                 }
    49.             }
    50.  
    51.         } catch (Exception e) {
    52.             Loggers.SRV_LOG.warn("Exception while processing client beat time out.", e);
    53.         }
    54.  
    55.     }

    到此完成删除实例的过程，整体的时序图如下:

    经过源码的研究，我发现问题的根本原因就是我的服务里面维持向nacos服务器端发送心跳的定时任务不再执行了，本质上讲 就是内存溢出的原因导致这个线程不再执行了

    应该从jvm和垃圾回收层面找问题,或者说JVM内存溢出造成的tomcat假死

  • 相关阅读:
    平衡搜索树——B-树小记
    1-k8s常见注意事项
    ＜蓝桥杯软件赛＞零基础备赛20周--第5周--杂题-2
    c++ | makefile | 编译 | 链接库
    2022-08-25 mysql范围查询场景下的自适应跳表索引-第二版-改进点
    zzcase&接口自动化-质&效的探索
    makefile & dockerfile
    virtio-net 报文组织形式
    Python 正则表达式转义
    R语言APRIORI关联规则、K-MEANS均值聚类分析中药专利复方治疗用药规律网络可视化...
  • 原文地址：https://blog.csdn.net/baidu_39322753/article/details/132804728