Alibaba：Nacos服务心跳和健康检查源码介绍

绝地灬酷狼 2022-10-14 00:57 126阅读 0赞

### 推荐阅读： ###

*  [**这套Github上40K+star学习笔记，可以帮你搞定95%以上的Java面试**][Github_40K_star_95_Java]
 *  [**毫不夸张的说，这份SpringBoot学习指南能解决你遇到的98%的问题**][Github_40K_star_95_Java]
 *  [**给跪了！这套万人期待的 SQL 成神之路PDF，终于开源了**][Github_40K_star_95_Java]

### 服务心跳 ###

Nacos Client会维护一个定时任务通过持续调用服务端的接口更新心跳时间，保证自己处于存活状态，防止服务端将服务剔除，Nacos默认5秒向服务端发送一次，通过请求服务端接口/instance/beat发送心跳。  
客户端服务在注册服务的时候会增加一个心跳的任务，如下图所示:

![50d4a835ec696b35a44b2cff086889a6.png][]

首先看下BeatInfo这个类，重点看标注的字段，该字段是给周期任务设定时间，如下图:

![8dad8b306339974a3bd4b258634268e4.png][]

该方法内部定义的一个DEFAULT\_HEART\_BEAT\_INTERVAL的常量，设定5秒:

![2151eb079ebee77051d0352d1d92ed73.png][]

接下来我们看下addBeatInfo方法，该方法内部主要是将BeatTask任务加入到线程池ScheduledExecutorService当中，如下图:

![9651f8df235d212eef0593207463af64.png][]

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

![bb9dea1680915112c322f8f70eac60ac.png][]

![078c52da0adcfe119fc3d6db2b4a702a.png][]

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

@CanDistro
    @PutMapping("/beat")
    @Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
    public ObjectNode beat(HttpServletRequest request) throws Exception {
    
        ObjectNode result = JacksonUtils.createEmptyJsonNode();
        //设置心跳间隔
        result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, switchDomain.getClientBeatInterval());
    
        String beat = WebUtils.optional(request, "beat", StringUtils.EMPTY);
        RsInfo clientBeat = null;
        //判断有无心跳内容
        //如果存在心跳内容则不是轻量级心跳就转化为RsInfo
        if (StringUtils.isNotBlank(beat)) {
            clientBeat = JacksonUtils.toObj(beat, RsInfo.class);
        }
        String clusterName = WebUtils
                .optional(request, CommonParams.CLUSTER_NAME, UtilsAndCommons.DEFAULT_CLUSTER_NAME);
        String ip = WebUtils.optional(request, "ip", StringUtils.EMPTY);
        int port = Integer.parseInt(WebUtils.optional(request, "port", "0"));
        if (clientBeat != null) {
            if (StringUtils.isNotBlank(clientBeat.getCluster())) {
                clusterName = clientBeat.getCluster();
            } else {
                // fix #2533
                clientBeat.setCluster(clusterName);
            }
            ip = clientBeat.getIp();
            port = clientBeat.getPort();
        }
        String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
        String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
        NamingUtils.checkServiceNameFormat(serviceName);
        Loggers.SRV_LOG.debug("[CLIENT-BEAT] full arguments: beat: {}, serviceName: {}", clientBeat, serviceName);
        //获取实例的信息
        Instance instance = serviceManager.getInstance(namespaceId, serviceName, clusterName, ip, port);
        //如果实例不存在
        if (instance == null) {
            if (clientBeat == null) {
                result.put(CommonParams.CODE, NamingResponseCode.RESOURCE_NOT_FOUND);
                return result;
            }
    
            Loggers.SRV_LOG.warn("[CLIENT-BEAT] The instance has been removed for health mechanism, "
                    + "perform data compensation operations, beat: {}, serviceName: {}", clientBeat, serviceName);
            //根据您心跳内容创建一个实例信息
            instance = new Instance();
            instance.setPort(clientBeat.getPort());
            instance.setIp(clientBeat.getIp());
            instance.setWeight(clientBeat.getWeight());
            instance.setMetadata(clientBeat.getMetadata());
            instance.setClusterName(clusterName);
            instance.setServiceName(serviceName);
            instance.setInstanceId(instance.getInstanceId());
            instance.setEphemeral(clientBeat.isEphemeral());
            //注册实例
            serviceManager.registerInstance(namespaceId, serviceName, instance);
        }
        //获取服务的信息
        Service service = serviceManager.getService(namespaceId, serviceName);
    
        if (service == null) {
            throw new NacosException(NacosException.SERVER_ERROR,
                    "service not found: " + serviceName + "@" + namespaceId);
        }
        //不存在的话，要创建一个进行处理
        if (clientBeat == null) {
            clientBeat = new RsInfo();
            clientBeat.setIp(ip);
            clientBeat.setPort(port);
            clientBeat.setCluster(clusterName);
        }
        //开启心跳检查任务
        service.processClientBeat(clientBeat);
    
        result.put(CommonParams.CODE, NamingResponseCode.OK);
        //5秒间隔
        if (instance.containsMetadata(PreservedMetadataKeys.HEART_BEAT_INTERVAL)) {
            result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, instance.getInstanceHeartBeatInterval());
        }
        //告诉客户端不需要带上心跳信息了，变成轻量级心跳了
        result.put(SwitchEntry.LIGHT_BEAT_ENABLED, switchDomain.isLightBeatEnabled());
        return result;
    }

接下来我们看一下processClientBeat方法，该方法将ClientBeatProcessor放入到线程池中，接下来我们看下重点看下run方法,

![f15d7914c206143a35720d673f7eeba7.png][]

该方法内部主要就是更新对应实例下心跳时间，整体上如下图:

![253781d8e20ae35391160b87301c6203.png][]

至此完成了从客户端到服务端更新实例的心跳时间，下图是整体的时序图:

![9b79a5720437aba4ac38ca490467c801.png][]

### 服务的健康检查 ###

Nacos Server会开启一个定时任务来检查注册服务的健康情况，对于超过15秒没收到客户端的心跳实例会将它的 healthy属性置为false，此时当客户端不会将该实例的信息发现，如果某个服务的实例超过30秒没收到心跳，则剔除该实例，如果剔除的实例恢复，发送心跳则会恢复。  
当有实例注册的时候，我们会看到有个service.init()的方法，该方法的实现主要是将ClientBeatCheckTask加入到线程池当中，如下图:

![6713f390a517555bb25f17ef8e46deed.png][]

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

public void run() {
        try {
            if (!getDistroMapper().responsible(service.getName())) {
                return;
            }
    
            if (!getSwitchDomain().isHealthCheckEnabled()) {
                return;
            }
            //获取服务所有实例信息
            List<Instance> instances = service.allIPs(true);
    
            // first set health status of instances:
            for (Instance instance : instances) {
                //如果心跳时间超过15秒则设置该实例信息为不健康状况
                if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) {
                    if (!instance.isMarked()) {
                        if (instance.isHealthy()) {
                            instance.setHealthy(false);
                            Loggers.EVT_LOG
                                    .info("{POS} {IP-DISABLED} valid: {}:{}@{}@{}, region: {}, msg: client timeout after {}, last beat: {}",
                                            instance.getIp(), instance.getPort(), instance.getClusterName(),
                                            service.getName(), UtilsAndCommons.LOCALHOST_SITE,
                                            instance.getInstanceHeartBeatTimeOut(), instance.getLastBeat());
                            getPushService().serviceChanged(service);
                            ApplicationUtils.publishEvent(new InstanceHeartbeatTimeoutEvent(this, instance));
                        }
                    }
                }
            }
    
            if (!getGlobalConfig().isExpireInstance()) {
                return;
            }
    
            // then remove obsolete instances:
            for (Instance instance : instances) {
    
                if (instance.isMarked()) {
                    continue;
                }
                //如果心跳时间超过30秒则删除该实例信息
                if (System.currentTimeMillis() - instance.getLastBeat() > instance.getIpDeleteTimeout()) {
                    // delete instance
                    Loggers.SRV_LOG.info("[AUTO-DELETE-IP] service: {}, ip: {}", service.getName(),
                            JacksonUtils.toJson(instance));
                    deleteIp(instance);
                }
            }
    
        } catch (Exception e) {
            Loggers.SRV_LOG.warn("Exception while processing client beat time out.", e);
        }
    
    }

首先我们来看一下deleteIp方法，该方法内部主要通过构建删除请求，发送删除请求，如下图:

![8f1cbe793bb61f351b8fbb7d97c1f032.png][]

删除实例的接口如下图:

![d4e984b55d8f29be98d8864780680c70.png][]

内部通过调用ServiceManager的removeInstance方法，如下图:

![7bb00553954b4b7ed21596a1a67c5796.png][]

重点看下substractIpAddresses内部通过调用updateIpAddresses，该方法内部主要就是移除到超过30秒的实例信息，如下图:

![8a8146e8a77b78bf8376a0f9b6ed3072.png][]

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

![2af6fc1d198d3a12b47a84c0d0966042.png][]

接下来我们看标记不健康时候的代码，这部分代码在客户端注册的时候也出现相同的代码，只是我们略过了，这部分也是观察者模式的重要体现，从这里我们可以学习到的东西在于结合Spring的事件机制，轻松实现观察者模式，当然这个里面也有部分我感觉写的不太好，哈哈，大佬们看到勿喷。

![1a6347b6b79a872f4d5cd1c3e77bbe3e.png][]

首先我们看serviceChanged方法，该方法主要是发布一个服务不健康的事件，如下图:

![9755b03d26c6ca80d1717a76dd510ec6.png][]

接下来我们看下如何处理这个事件，这个时候涉及PushService这个类，整体的继承结构如下图:

![a3d2dd0a1dd7af1ba9d9f44db54d6354.png][]

我们看到该类的继承ApplicationListener接口，该接口是一个支持泛型的接口，传入了ServiceChangeEvent的类，此处就是对事件的处理，如下图:

![64e99ad8f99aeb9eb0e364cc3595e957.png][]

接下来看一下onApplicationEvent方法，这个方法主要完成了准备数据，发送数据这几件事情:

public void onApplicationEvent(ServiceChangeEvent event) {
        Service service = event.getService();
        String serviceName = service.getName();
        String namespaceId = service.getNamespaceId();
    
        Future future = GlobalExecutor.scheduleUdpSender(() -> {
            try {
                Loggers.PUSH.info(serviceName + " is changed, add it to push queue.");
                //获取所有需要推送的客户端
                ConcurrentMap<String, PushClient> clients = clientMap
                        .get(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName));
                if (MapUtils.isEmpty(clients)) {
                    return;
                }
    
                Map<String, Object> cache = new HashMap<>(16);
                long lastRefTime = System.nanoTime();
                for (PushClient client : clients.values()) {
                    //超时的不删除跳过处理
                    if (client.zombie()) {
                        Loggers.PUSH.debug("client is zombie: " + client.toString());
                        clients.remove(client.toString());
                        Loggers.PUSH.debug("client is zombie: " + client.toString());
                        continue;
                    }
    
                    Receiver.AckEntry ackEntry;
                    Loggers.PUSH.debug("push serviceName: {} to client: {}", serviceName, client.toString());
                    String key = getPushCacheKey(serviceName, client.getIp(), client.getAgent());
                    byte[] compressData = null;
                    Map<String, Object> data = null;
    
                    if (switchDomain.getDefaultPushCacheMillis() >= 20000 && cache.containsKey(key)) {
                        org.javatuples.Pair pair = (org.javatuples.Pair) cache.get(key);
                        compressData = (byte[]) (pair.getValue0());
                        data = (Map<String, Object>) pair.getValue1();
                        Loggers.PUSH.debug("[PUSH-CACHE] cache hit: {}:{}", serviceName, client.getAddrStr());
                    }
                    //准备UDP数据
                    if (compressData != null) {
                        ackEntry = prepareAckEntry(client, compressData, data, lastRefTime);
                    } else {
                        ackEntry = prepareAckEntry(client, prepareHostsData(client), lastRefTime);
                        if (ackEntry != null) {
                            cache.put(key, new org.javatuples.Pair<>(ackEntry.origin.getData(), ackEntry.data));
                        }
                    }
    
                    Loggers.PUSH.info("serviceName: {} changed, schedule push for: {}, agent: {}, key: {}",
                            client.getServiceName(), client.getAddrStr(), client.getAgent(),
                            (ackEntry == null ? null : ackEntry.key));
                    //发送数据
                    udpPush(ackEntry);
                }
            } catch (Exception e) {
                Loggers.PUSH.error("[NACOS-PUSH] failed to push serviceName: {} to client, error: {}", serviceName, e);
    
            } finally {
                //发送完成删除
                futureMap.remove(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName));
            }
    
        }, 1000, TimeUnit.MILLISECONDS);
        //增加待推送的任务
        futureMap.put(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName), future);
    
    }

接下里我们重点看下udpPush的方法，整个方法主要是通过一个Map对象来记录UDP请求，如果没收到就重试发送请求，整体如下:

private static Receiver.AckEntry udpPush(Receiver.AckEntry ackEntry) {
        if (ackEntry == null) {
            Loggers.PUSH.error("[NACOS-PUSH] ackEntry is null.");
            return null;
        }
    
        //如果大于最大的尝试次数
        //移除发送的数据和待确认的key
        //失败推送的次数+1
        if (ackEntry.getRetryTimes() > MAX_RETRY_TIMES) {
            Loggers.PUSH.warn("max re-push times reached, retry times {}, key: {}", ackEntry.retryTimes, ackEntry.key);
            ackMap.remove(ackEntry.key);
            udpSendTimeMap.remove(ackEntry.key);
            failedPush += 1;
            return ackEntry;
        }
    
        try {
            if (!ackMap.containsKey(ackEntry.key)) {
                totalPush++;
            }
            //记录UDP请求的返回信息
            ackMap.put(ackEntry.key, ackEntry);
            udpSendTimeMap.put(ackEntry.key, System.currentTimeMillis());
    
            Loggers.PUSH.info("send udp packet: " + ackEntry.key);
            //发送UDP请求
            udpSocket.send(ackEntry.origin);
    
            ackEntry.increaseRetryTime();
            //如果UDP没收到返回信息 每10秒尝试一下
            GlobalExecutor.scheduleRetransmitter(new Retransmitter(ackEntry),
                    TimeUnit.NANOSECONDS.toMillis(ACK_TIMEOUT_NANOS), TimeUnit.MILLISECONDS);
    
            return ackEntry;
        } catch (Exception e) {
            Loggers.PUSH.error("[NACOS-PUSH] failed to push data: {} to client: {}, error: {}", ackEntry.data,
                    ackEntry.origin.getAddress().getHostAddress(), e);
            ackMap.remove(ackEntry.key);
            udpSendTimeMap.remove(ackEntry.key);
            failedPush += 1;
    
            return null;
        }
    }

\} catch (Exception e) \{  
Loggers.PUSH.error("\[NACOS-PUSH\] failed to push data: \{\} to client: \{\}, error: \{\}", ackEntry.data,  
ackEntry.origin.getAddress().getHostAddress(), e);  
ackMap.remove(ackEntry.key);  
udpSendTimeMap.remove(ackEntry.key);  
failedPush += 1;

return null;
    }

[Github_40K_star_95_Java]: https://docs.qq.com/doc/DSmxTbFJ1cmN1R2dB
[50d4a835ec696b35a44b2cff086889a6.png]: /images/20221014/15a7a6662fef43a2a76431f8be04d3c1.png
[8dad8b306339974a3bd4b258634268e4.png]: /images/20221014/abd381d5ef9141719ae12c7e6f974b95.png
[2151eb079ebee77051d0352d1d92ed73.png]: /images/20221014/1ab81251227040ceaf2bdfa1365eadf2.png
[9651f8df235d212eef0593207463af64.png]: /images/20221014/fa64102ee4a345478b0706999b840a8f.png
[bb9dea1680915112c322f8f70eac60ac.png]: /images/20221014/4e96205812a9487ea1c69f678832a0ff.png
[078c52da0adcfe119fc3d6db2b4a702a.png]: /images/20221014/1fd64031c33a4b4f87a3ce240c2a57b9.png
[f15d7914c206143a35720d673f7eeba7.png]: /images/20221014/ba77c08dfe5a4b55ae6ac99019105f4e.png
[253781d8e20ae35391160b87301c6203.png]: /images/20221014/c11e6198243442cdaa98caab5122c742.png
[9b79a5720437aba4ac38ca490467c801.png]: /images/20221014/5443631691b14759906eacfd85562c46.png
[6713f390a517555bb25f17ef8e46deed.png]: /images/20221014/e036a02d562142a59c32edf5d0545852.png
[8f1cbe793bb61f351b8fbb7d97c1f032.png]: /images/20221014/e7d697dc4a9448e09909b995ae6b25db.png
[d4e984b55d8f29be98d8864780680c70.png]: /images/20221014/761c243d8054442aa6a39b2a43851d2f.png
[7bb00553954b4b7ed21596a1a67c5796.png]: /images/20221014/2ef3a72eb72a466f8ec22182b71734c4.png
[8a8146e8a77b78bf8376a0f9b6ed3072.png]: /images/20221014/ea4795e150694a508b3de876e34c2a3e.png
[2af6fc1d198d3a12b47a84c0d0966042.png]: /images/20221014/cdf1be39c90b4b5794fc8b8480716ce4.png
[1a6347b6b79a872f4d5cd1c3e77bbe3e.png]: /images/20221014/25b2f0dff57f4afdab60901b33af9c5b.png
[9755b03d26c6ca80d1717a76dd510ec6.png]: /images/20221014/4a35265824b34e56bcd01f651b057331.png
[a3d2dd0a1dd7af1ba9d9f44db54d6354.png]: /images/20221014/e7a3451a74754e2d949d39cb57e246c5.png
[64e99ad8f99aeb9eb0e364cc3595e957.png]: /images/20221014/59da0cee676845eb860c80badca269b0.png