Nacos注册中心10-Server端（处理服务心跳请求）

0. 环境

• nacos版本：1.4.1
• Spring Cloud : 2020.0.2
• Spring Boot ：2.4.4
• Spring Cloud alibaba: 2.2.5.RELEASE

测试代码：github.com/hsfxuebao/s…

1. 服务端接收心跳

这个心跳请求是走了InstanceController 的beat 方法处理的，代码如下:

@CanDistro
@PutMapping("/beat")
@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
public ObjectNode beat(HttpServletRequest request) throws Exception {
    // 创建一个JSON Node，该方法的返回值就是它，后面的代码就是对这个Node进行各种初始化
    ObjectNode result = JacksonUtils.createEmptyJsonNode();
    result.put(SwitchEntry.CLIENT_BEAT_INTERVAL, switchDomain.getClientBeatInterval());
 
    // 从请求中获取到beat，即client端的beatInfo
    String beat = WebUtils.optional(request, "beat", StringUtils.EMPTY);
    RsInfo clientBeat = null;
    // 将beat构建为clientBeat
    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);
    // 获取到客户端传递来的client的port，其将来用于UDP通信
    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);
    // 从注册表中获取当前发送请求的client对应的instance
    Instance instance = serviceManager.getInstance(namespaceId, serviceName, clusterName, ip, port);
    // 处理注册表中不存在该client的instance的情况
    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);
        // 下面处理的情况是，注册表中没有该client的instance，但其发送的请求中具有心跳数据。
        // 在client的注册请求还未到达时（网络抖动等原因），第一次心跳请求先到达了server，会出现这种情况
        // 处理方式是，使用心跳数据构建出一个instance，注册到注册表
        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 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);
    }
    // todo 处理本次心跳
    service.processClientBeat(clientBeat);
 
    result.put(CommonParams.CODE, NamingResponseCode.OK);
 
    // 这个就有点动态配置了
    // 如果instance中有 preserved.heart.beat.interval 这个参数
    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;
}
复制代码

先是根据namespaceId, serviceName, clusterName, ip, port 这个参数调用 ServiceManager的getInstance 获取对应的instance，其实就是先根据namespace从serviceMap中获取对应的service，接着根据cluster从service的clusterMap中获取对应cluster的instance集合，然后再遍历比对ip与port。

如果没有找到对应的instance，而且beatInfo不是null，就会进行服务注册。

接着就是根据namespace与serviceName获取service，然后调用service的processClientBeat 方法处理心跳。这个processClientBeat 方法我们后面看，先看下后面这个有意思的，它往这个返回值中塞了clientBeatInterval 与lightBeatEnabled 参数值，这clientBeatInterval 就是心跳间隔，lightBeatEnabled 就是带不带beatInfo，这时候lightBeatEnabled 返回的就是true了，也就是下次不带了，看来这个心跳间隔是可以随时调整的，而且不用动服务，在控制台修改下某个实例的元数据就可以了。

接下来看下service是怎样处理请求的：

public void processClientBeat(final RsInfo rsInfo) {
    // 创建一个处理器，其是一个任务
    ClientBeatProcessor clientBeatProcessor = new ClientBeatProcessor();
    clientBeatProcessor.setService(this);
    clientBeatProcessor.setRsInfo(rsInfo);
    // 开启一个立即执行的任务，即执行clientBeatProcessor任务的run()
    HealthCheckReactor.scheduleNow(clientBeatProcessor);
}
复制代码

封装一个ClientBeatProcessor ，然后交给了HealthCheckReactor 的scheduleNamingHealth 方法，其实就是给了一个健康检查的线程池处理了。看下ClientBeatProcessor 这个任务里面怎样执行的:

@Override
public void run() {
    Service service = this.service;
    if (Loggers.EVT_LOG.isDebugEnabled()) {
        Loggers.EVT_LOG.debug("[CLIENT-BEAT] processing beat: {}", rsInfo.toString());
    }
 
    String ip = rsInfo.getIp();
    String clusterName = rsInfo.getCluster();
    int port = rsInfo.getPort();
    Cluster cluster = service.getClusterMap().get(clusterName);
    // 获取当前服务的所有临时实例
    List<Instance> instances = cluster.allIPs(true);
    // 遍历所有这些临时实例，从中查找当前发送心跳的instance
    for (Instance instance : instances) {
        // 只要ip与port与当前心跳的instance的相同，就是了
        if (instance.getIp().equals(ip) && instance.getPort() == port) {
            if (Loggers.EVT_LOG.isDebugEnabled()) {
                Loggers.EVT_LOG.debug("[CLIENT-BEAT] refresh beat: {}", rsInfo.toString());
            }
            // 修改最后心跳时间戳
            instance.setLastBeat(System.currentTimeMillis());
            // 修改该instance的健康状态
            // 当instance被标记时，即其marked为true时，其是一个持久实例
            if (!instance.isMarked()) {
                // instance的healthy才是临时实例健康状态的表示
                // 若当前instance健康状态为false，但本次是其发送的心跳，说明这个instance“起死回生”了，
                // 我们需要将其health变为true
                if (!instance.isHealthy()) {
                    instance.setHealthy(true);
                    Loggers.EVT_LOG
                            .info("service: {} {POS} {IP-ENABLED} valid: {}:{}@{}, region: {}, msg: client beat ok",
                                    cluster.getService().getName(), ip, port, cluster.getName(),
                                    UtilsAndCommons.LOCALHOST_SITE);
                    // todo 发布服务变更事件（其对后续我们要分析的UDP通信非常重要）
                    getPushService().serviceChanged(service);
                }
            }
        }
    }
}
复制代码

其实就是通过namespace/serviceName/cluster/ip/port找到对应的instance对象，重新设置一下LastBeat 的时间，也就是

instance.setLastBeat(System.currentTimeMillis());
复制代码

这行，接着就是判断，如果不健康的话，就更改健康状态是true，也就是改成健康。最后getPushService().serviceChanged(service);这行需要注意下，健康状态改变了，会引起它 将新的instance信息推送到那堆服务订阅者客户端上，这个服务订阅发布我们后面会介绍。

好了，到这我们服务端对心跳消息的处理就结束了，可以看到，处理心跳消息也是异步的，将处理封装成task投寄到线程池，然后就直接返回给客户端了，由线程池执行这个task。

2. 方法调用图