java语言中Nacos注册中心Server端实现处理服务主动下线请求

前言

Nacos注册中心的Server的处理方式

1. 环境

• 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…

2. 服务主动下线

客户端请求会被InstanceController 这个controller处理，其实与instance有关的请求都是这个controller 来处理。InstanceController的deregister这个方法就是服务下线。

@CanDistro
@DeleteMapping
@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
public String deregister(HttpServletRequest request) throws Exception {
    // 从请求中获取要操作的instance
    Instance instance = getIpAddress(request);
    String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
    String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
    NamingUtils.checkServiceNameFormat(serviceName);
    // 从注册表中获取service
    Service service = serviceManager.getService(namespaceId, serviceName);
    if (service == null) {
        Loggers.SRV_LOG.warn("remove instance from non-exist service: {}", serviceName);
        return "ok";
    }
    // todo 删除instance
    serviceManager.removeInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
    return "ok";
}

前面代码都是解析请求参数的，我们需要关注serviceManager.getService 与serviceManager.removeInstance 这两个行代码的调用，serviceManager.getService 其实就是根据namespace与serviceName 从serviceMap 这个map中获取对应的service 实例，如果没有的话，就说明没有之前没有注册过，也就直接返回ok了，如果存在的话，就会调用serviceManager.removeInstance 移除这个instance。

public void removeInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)
        throws NacosException {
    // 从注册表获取当前service
    Service service = getService(namespaceId, serviceName);

    synchronized (service) {
        // todo 删除
        removeInstance(namespaceId, serviceName, ephemeral, service, ips);
    }
}

这个方法没啥好看的，先获取一下这个namespace与serviceName对应的service实例，然后加锁，调用removeInstance 方法进行移除。

private void removeInstance(String namespaceId, String serviceName, boolean ephemeral, Service service,
        Instance... ips) throws NacosException {

    String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);
    // todo 从注册表中删除instance，返回下线完剩下的instance集合
    List<Instance> instanceList = substractIpAddresses(service, ephemeral, ips);

    Instances instances = new Instances();
    instances.setInstanceList(instanceList);
    // todo 将本次变更同步给其它nacos,交给一致性服务进行存储，通知等
    consistencyService.put(key, instances);
}

先生成一个服务列表的key这个key与你instance是否是临时节点有关系，如果是临时节点，生成的key是这个样子的com.alibaba.nacos.naming.iplist.ephemeral.{namespace}##{serviceName} 永久节点就是com.alibaba.nacos.naming.iplist.{namespace}##{serviceName} 这个样子。 接着就是调用substractIpAddresses 方法用之前的instance列表减去 这次要下线的实例列表，然后生成一份新的删除下线的实例列表。

private List<Instance> substractIpAddresses(Service service, boolean ephemeral, Instance... ips)
        throws NacosException {
    return updateIpAddresses(service, UtilsAndCommons.UPDATE_INSTANCE_ACTION_REMOVE, ephemeral, ips);
}

UPDATE_INSTANCE_ACTION_REMOVE这个action是remove。接着调用updateIpAddresses 方法:

public List<Instance> updateIpAddresses(Service service, String action, boolean ephemeral, Instance... ips)
        throws NacosException {
    // 从其它nacos获取当前服务数据（临时实例数据）
    Datum datum = consistencyService
            .get(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), ephemeral));
    // 获取本地注册表中当前服务的所有临时实例
    List<Instance> currentIPs = service.allIPs(ephemeral);
    Map<String, Instance> currentInstances = new HashMap<>(currentIPs.size());
    Set<String> currentInstanceIds = Sets.newHashSet();
    // 遍历注册表中获取到的实例
    for (Instance instance : currentIPs) {
        // 将当前遍历的instance写入到map，key为ip:port，value为instance
        currentInstances.put(instance.toIpAddr(), instance);
        // 将当前遍历的instanceId写入到一个set
        currentInstanceIds.add(instance.getInstanceId());
    }

    Map<String, Instance> instanceMap;
    if (datum != null && null != datum.value) {
        // todo 将注册表中主机的instance数据替换掉外来的相同主机的instance数据
        instanceMap = setValid(((Instances) datum.value).getInstanceList(), currentInstances);
    } else {
        instanceMap = new HashMap<>(ips.length);
    }

    for (Instance instance : ips) {
        // 若当前service中不包含当前要注册的instance所属cluster，则创建一个
        if (!service.getClusterMap().containsKey(instance.getClusterName())) {
            Cluster cluster = new Cluster(instance.getClusterName(), service);
            // todo 初始化cluster的健康检测任务
            cluster.init();
            service.getClusterMap().put(instance.getClusterName(), cluster);
            Loggers.SRV_LOG
                    .warn("cluster: {} not found, ip: {}, will create new cluster with default configuration.",
                            instance.getClusterName(), instance.toJson());
        }

        // 若当前操作为清除操作，则将当前instance从instanceMap中清除，
        // 否则就是添加操作，即将当前instance添加到instanceMap中
        if (UtilsAndCommons.UPDATE_INSTANCE_ACTION_REMOVE.equals(action)) {
            instanceMap.remove(instance.getDatumKey());
        } else {
            Instance oldInstance = instanceMap.get(instance.getDatumKey());
            if (oldInstance != null) {
                instance.setInstanceId(oldInstance.getInstanceId());
            } else {
                instance.setInstanceId(instance.generateInstanceId(currentInstanceIds));
            }
            instanceMap.put(instance.getDatumKey(), instance);
        }

    }

    if (instanceMap.size() <= 0 && UtilsAndCommons.UPDATE_INSTANCE_ACTION_ADD.equals(action)) {
        throw new IllegalArgumentException(
                "ip list can not be empty, service: "   service.getName()   ", ip list: "   JacksonUtils
                        .toJson(instanceMap.values()));
    }

    return new ArrayList<>(instanceMap.values());
}

其实就是将之前的instance弄出来，然后放到这个instanceMap中，然后遍历这个要删除的instance集合，如果是删除action的话，就从instanceMap中移除这个DatumKey，这个key就是这个样子 ip:port:unknown:{cluster}。到最后这个instanceMap就剩下抛去我们要下线的instance了。

接着removeInstance 这个方法往下看，就是创建一个instances对象，然后将instance集合塞到这里面。接着就是交给consistencyService组件来进行put操作。这里其实就是调用了EphemeralConsistencyService的实现类DistroConsistencyServiceImpl 的put方法。往下的步骤我们就不赘述了，再往下就与服务注册后续的逻辑一摸一样的，就是封装instance集合与key ，封装成一个Datum，然后将这个Datum 塞到dataStore 这个存储组件中，这个组件实际就是个map。

接着就是往Notifier 这个组件里面的一个task队列添加一个事件通知任务，然后就完事了（其实这里还有向其他server 同步的步骤，我们这里暂时先不研究），就可以将响应返回给客户端了。这个时候，其实service 对象里面的instance列表并没有更新，这就是所谓nacos的异步注册异步下线，会有一个后台线程，不停的从Notifier组件中的task 队列中取出task，然后调用handle方法进行事件通知，其实就通知到service 对象的onChange 方法里面了，其实更新操作都是这个方法做的。具体的源码分析可以看下我们这里也就不再赘述了，然后好好理解下这个nacos所谓的异步注册是怎样实现的。

3. 方法调用图

这篇好文章是转载于：学新通技术网