java语言中Nacos注册中心：Server端【处理注册请求】

前言

java语言中Nacos注册中心：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#register

nacos注册中心功能是在naming这个子项目下面的，是个springboot项目，然后我们直接找controller就可以了，找到这个InstanceController，第一个方法就是服务注册方法（register方法）:

@CanDistro
@PostMapping
@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
public String register(HttpServletRequest request) throws Exception {
    // 从请求中获取指定属性值
    final String namespaceId = WebUtils
            .optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
    // 从请求中获取指定属性值
    final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
    // 检测serviceName是否合法
    NamingUtils.checkServiceNameFormat(serviceName);

    // 通过请求参数组装出instance
    final Instance instance = parseInstance(request);

    // todo 将instance写到注册表
    serviceManager.registerInstance(namespaceId, serviceName, instance);
    return "ok";
}

先是解析出来instance，就是根据client发送的那堆参数解析出来的。接着就是调用serviceManager组件进行实例注册，这个serviceManager 组件在注册中心是个核心组件（服务注册，下线，获取服务列表等），都是找这个组件的。

2. ServiceManager#registerInstance

public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {

    // 创建 空service
    // 第三个参数 true表示临时实例
    createEmptyService(namespaceId, serviceName, instance.isEphemeral());

    // 从注册表获取到service
    Service service = getService(namespaceId, serviceName);
    // 这里指定不能为null
    if (service == null) {
        throw new NacosException(NacosException.INVALID_PARAM,
                "service not found, namespace: "   namespaceId   ", service: "   serviceName);
    }
    // todo 将instance写入到service，即写入到了注册表
    addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}

这几个步骤都很重要，我们挨个看看，createEmptyService这个方法就是当service 不存在的时候，创建一个空的serivce，这service你可以理解为服务，其实就是服务的意思，看下这个方法

2.1 ServiceManager#createEmptyService

public void createEmptyService(String namespaceId, String serviceName, boolean local) throws NacosException {
    // local为true，表示当前实例为临时实例
    createServiceIfAbsent(namespaceId, serviceName, local, null);
}

public void createServiceIfAbsent(String namespaceId, String serviceName, boolean local, Cluster cluster)
        throws NacosException {
    // 从注册表中获取service
    Service service = getService(namespaceId, serviceName);
    // 若当前注册instance是其提供服务的第一个实例，则注册表中是没有该service的，
    // 此时会创建一个service实例
    if (service == null) {

        Loggers.SRV_LOG.info("creating empty service {}:{}", namespaceId, serviceName);
        service = new Service();
        service.setName(serviceName);
        service.setNamespaceId(namespaceId);
        service.setGroupName(NamingUtils.getGroupName(serviceName));
        // now validate the service. if failed, exception will be thrown
        // 修改时间
        service.setLastModifiedMillis(System.currentTimeMillis());
        // todo 重新计算校验和
        service.recalculateChecksum();
        if (cluster != null) {
            // cluster与service建立联系
            cluster.setService(service);
            service.getClusterMap().put(cluster.getName(), cluster);
        }
        service.validate();
        // todo 将service写入到注册表
        putServiceAndInit(service);
        // 对持久实例的操作
        if (!local) {
            addOrReplaceService(service);
        }
    }
}

先是根据namespace与serviceName 获取service ，如果没有的话，就创建，最开始的时候，肯定是没有的，然后就会创建一个service，看下这个getService 方法:

public Service getService(String namespaceId, String serviceName) {
    if (serviceMap.get(namespaceId) == null) {
        return null;
    }
    return chooseServiceMap(namespaceId).get(serviceName);
}

说白了其实就是去serviceMap 这个成员中获取，对应关系看下面这个注释就可以。

// namespace   ---》 serivceName ，service
private final Map<String, Map<String, Service>> serviceMap = new ConcurrentHashMap<>();

在createEmptyService 这个方法，最后putServiceAndInit(service);这行代码也需要注意下

private void putServiceAndInit(Service service) throws NacosException {
    // 将service写入注册表
    putService(service);
    // 初始化service内部健康检测任务
    service.init();
    // 给nacos集合中的当前服务的持久实例、临时实例添加监听
    consistencyService
            .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true), service);
    consistencyService
            .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), false), service);
    Loggers.SRV_LOG.info("[NEW-SERVICE] {}", service.toJson());
}

就是将这个service 放到 serviceMap中，然后service进行初始化， 再就是添加两个监听器。 看下这个service 初始化的方法。

public void init() {
    // 开启定时清除过期instance任务
    HealthCheckReactor.scheduleCheck(clientBeatCheckTask);
    // 开启了当前service所包含的所有cluster的健康检测任务
    for (Map.Entry<String, Cluster> entry : clusterMap.entrySet()) {
        entry.getValue().setService(this);
        // 开启当前遍历cluster的健康检测任务：
        // 将当前cluster包含的所有instance的心跳检测任务定时添加到一个任务队列
        // taskQueue，即将当前cluster所包含的持久实例的心跳任务添加到taskQueue
        entry.getValue().init();
    }
}

这里这个初始化有个非常重要的地方就是往健康检查器中添加一个任务，健康检查的任务，这个任务其实就是扫描这个service里面长时间没有心跳的instance（服务实例），然后进行健康状态改变，服务下线。

接着回到ServiceManager#registerInstance 方法中去

public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {

    // 创建 空service
    // 第三个参数 true表示临时实例
    createEmptyService(namespaceId, serviceName, instance.isEphemeral());

    // 从注册表获取到service
    Service service = getService(namespaceId, serviceName);
    // 这里指定不能为null
    if (service == null) {
        throw new NacosException(NacosException.INVALID_PARAM,
                "service not found, namespace: "   namespaceId   ", service: "   serviceName);
    }
    // todo 将instance写入到service，即写入到了注册表
    addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}

这个时候再获取service就能够获取到了，然后接着就是调用addInstance 方法添加实例了。

public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)
        throws NacosException {

    String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);
    // 从注册表中获取service
    Service service = getService(namespaceId, serviceName);

    synchronized (service) {
        // todo 将要注册的instance写入到service，即写入到了注册表
        List<Instance> instanceList = addIpAddresses(service, ephemeral, ips);

        // 塞到instance
        Instances instances = new Instances();
        instances.setInstanceList(instanceList);
        // todo 将本次变更同步给其它Nacos
        consistencyService.put(key, instances);
    }
}

首先是生成一个key，这个key是根据你namespace，serviceName ，是否临时节点来决定的，我们这里是临时的，直接看下生成的结果就可以了

com.alibaba.nacos.naming.iplist.ephemeral.{namespace}##{serviceName}

接着就是获取service， 上锁，调用addIpAddresses 得到一个instance集合。

private List<Instance> addIpAddresses(Service service, boolean ephemeral, Instance... ips) throws NacosException {
    // 修改当前service的instance列表，这个修改一共有两种操作：
    // 添加实例  与  删除实例
    return updateIpAddresses(service, UtilsAndCommons.UPDATE_INSTANCE_ACTION_ADD, ephemeral, ips);
}

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

UPDATE_INSTANCE_ACTION_ADD是add，主要就是新的instance 与之前的instance进行合并啥的，生成一个新的instance集合。

接着就是创建一个instances 对象，将instance集合塞进去，最后调用consiitencyService 组件进行保存。

3. DelegateConsistencyServiceImpl#put

@Override
public void put(String key, Record value) throws NacosException {
    mapConsistencyService(key).put(key, value);
}

这个方法会根据你这个key是临时的还是永久的选择一个consisitencyService:

private ConsistencyService mapConsistencyService(String key) {
    // 判断是不是临时key 临时key就走ephemeralConsistencyService服务，否则就走persistentConsistencyService服务
    return KeyBuilder.matchEphemeralKey(key) ? ephemeralConsistencyService : persistentConsistencyService;
}

这里我们是临时的，所以就走EphemeralConsistencyService 的实现类DistroConsistencyServiceImpl 的put方法。

3.1 DistroConsistencyServiceImpl#put

// put 往存储器中放入数据
@Override
public void put(String key, Record value) throws NacosException {
    // 往本地里面存储
    onPut(key, value);
    // todo 应该是同步
    distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,
            globalConfig.getTaskDispatchPeriod() / 2);
}

我们直接看onPut方法就可以了，后面这个方法我们现在先不研究:

public void onPut(String key, Record value) {

    // 判断是否是临时
    if (KeyBuilder.matchEphemeralInstanceListKey(key)) {
        // 封装到datnum
        Datum<Instances> datum = new Datum<>();
        datum.value = (Instances) value;
        datum.key = key;
        // 自增
        datum.timestamp.incrementAndGet();
        // dataStore存储
        dataStore.put(key, datum);
    }

    // 如果listener里面没有这个key 直接返回，发布订阅，进行通知，这个监听器会在Service初始化的时候添加进去
    if (!listeners.containsKey(key)) {
        return;
    }

    // 添加通知任务
    notifier.addTask(key, DataOperation.CHANGE);
}

这里需要判断一下是否是临时节点，如果是的话，就封装一个Datum，这个东西就是个kv，不用太多care它，接着就是调用datastore 的put方法进行存储。

最后如果有这个key的监听器的话，就会接着往下走，没有的话就返回，我们在初始化service 的时候是注册了2个监听器的，往上翻翻就可以看到。接着就是调用notifier添加任务。

先看下这个datastore:

private Map<String, Datum> dataMap = new ConcurrentHashMap<>(1024);

public void put(String key, Datum value) {
    dataMap.put(key, value);
}

这个特别简单，就是个map，然后往map里面塞东西，接着看下这个notifier.addTask(key, DataOperation.CHANGE);

// 添加通知任务到队列中
public void addTask(String datumKey, DataOperation action) {

    // 如果已经存在，并且是change事件
    if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {
        return;
    }
    if (action == DataOperation.CHANGE) {
        // 往services缓存中放
        services.put(datumKey, StringUtils.EMPTY);
    }
    // 加入任务队列中
    tasks.offer(Pair.with(datumKey, action));
}

private BlockingQueue<Pair<String, DataOperation>> tasks = new ArrayBlockingQueue<>(1024 * 1024);

这里其实就是往任务队列中添加了一个任务。到这按理说我们服务注册就该结束了，但是，我们发现生成了新的instance集合并没有更新到service对象里面去，所以还得继续往下看，看看这个通知任务是怎么回事。 其实DistroConsistencyServiceImpl 这个类在初始化的时候，然后提交了一个任务:

// 初始化操作
@PostConstruct
public void init() {
    GlobalExecutor.submitDistroNotifyTask(notifier);
}

我们看下这个Notifier 的run方法

@Override
public void run() {
    Loggers.DISTRO.info("distro notifier started");

    for (; ; ) {
        try {
            // 取出任务
            Pair<String, DataOperation> pair = tasks.take();
            // todo 任务处理
            handle(pair);
        } catch (Throwable e) {
            Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);
        }
    }
}

就是取出任务队列中的任务，调用handle方法进行处理:

private void handle(Pair<String, DataOperation> pair) {
        try {
            String datumKey = pair.getValue0();
            DataOperation action = pair.getValue1();

            // 从services缓存中移除
            services.remove(datumKey);

            int count = 0;

            if (!listeners.containsKey(datumKey)) {
                return;
            }

            for (RecordListener listener : listeners.get(datumKey)) {

                count  ;

                try {
                    // 通知数据已经改变
                    if (action == DataOperation.CHANGE) {
                        // todo 将key对应的实例列表传过去
                        listener.onChange(datumKey, dataStore.get(datumKey).value);
                        continue;
                    }

                    // 通知数据已经删除
                    if (action == DataOperation.DELETE) {
                        listener.onDelete(datumKey);
                        continue;
                    }
                } catch (Throwable e) {
                    Loggers.DISTRO.error("[NACOS-DISTRO] error while notifying listener of key: {}", datumKey, e);
                }
            }

            if (Loggers.DISTRO.isDebugEnabled()) {
                Loggers.DISTRO
                        .debug("[NACOS-DISTRO] datum change notified, key: {}, listener count: {}, action: {}",
                                datumKey, count, action.name());
            }
        } catch (Throwable e) {
            Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);
        }
    }
}

这里直接通知，调用listener 的onChange或onDelete执行相关的工作。我们那个时候是将service作为listener注册进来了，所以我们看下service的onChange方法:

public void onChange(String key, Instances value) throws Exception {

    Loggers.SRV_LOG.info("[NACOS-RAFT] datum is changed, key: {}, value: {}", key, value);

    for (Instance instance : value.getInstanceList()) {

        if (instance == null) {
            // Reject this abnormal instance list:
            throw new RuntimeException("got null instance "   key);
        }

        // 处理权重问题
        if (instance.getWeight() > 10000.0D) {
            instance.setWeight(10000.0D);
        }

        if (instance.getWeight() < 0.01D && instance.getWeight() > 0.0D) {
            instance.setWeight(0.01D);
        }
    }

    // todo 更新instance
    updateIPs(value.getInstanceList(), KeyBuilder.matchEphemeralInstanceListKey(key));

    recalculateChecksum();
}

核心的在updateIPs这个方法方法中，参数1是新的instance集合，参数2是是否是临时节点:

public void updateIPs(Collection<Instance> instances, boolean ephemeral) {
    Map<String, List<Instance>> ipMap = new HashMap<>(clusterMap.size());
    for (String clusterName : clusterMap.keySet()) {
        ipMap.put(clusterName, new ArrayList<>());
    }

    for (Instance instance : instances) {
        try {
            if (instance == null) {
                Loggers.SRV_LOG.error("[NACOS-DOM] received malformed ip: null");
                continue;
            }

            // cluster为null 就设置成默认DEFAULT
            if (StringUtils.isEmpty(instance.getClusterName())) {
                instance.setClusterName(UtilsAndCommons.DEFAULT_CLUSTER_NAME);
            }

            // cluster不存在就创建对应的cluster
            if (!clusterMap.containsKey(instance.getClusterName())) {
                Loggers.SRV_LOG
                        .warn("cluster: {} not found, ip: {}, will create new cluster with default configuration.",
                                instance.getClusterName(), instance.toJson());
                Cluster cluster = new Cluster(instance.getClusterName(), this);
                cluster.init();
                // 加入map中
                getClusterMap().put(instance.getClusterName(), cluster);
            }

            List<Instance> clusterIPs = ipMap.get(instance.getClusterName());
            if (clusterIPs == null) {
                clusterIPs = new LinkedList<>();
                ipMap.put(instance.getClusterName(), clusterIPs);
            }

            clusterIPs.add(instance);
        } catch (Exception e) {
            Loggers.SRV_LOG.error("[NACOS-DOM] failed to process ip: "   instance, e);
        }
    }

    for (Map.Entry<String, List<Instance>> entry : ipMap.entrySet()) {
        //make every ip mine
        List<Instance> entryIPs = entry.getValue();
        // 某个集群，更新下面的instance
        clusterMap.get(entry.getKey()).updateIps(entryIPs, ephemeral);
    }

    // 设置 最近的一次修改时间
    setLastModifiedMillis(System.currentTimeMillis());
    // 获取pushService 然后服务改变，通知改变
    getPushService().serviceChanged(this);
    StringBuilder stringBuilder = new StringBuilder();

    for (Instance instance : allIPs()) {
        stringBuilder.append(instance.toIpAddr()).append("_").append(instance.isHealthy()).append(",");
    }

    Loggers.EVT_LOG.info("[IP-UPDATED] namespace: {}, service: {}, ips: {}", getNamespaceId(), getName(),
            stringBuilder.toString());

}

这个方法其实就是遍历instance集合，然后更新clusterMap 这个里面的内容，这个clusterMap 其实就是clusterName 与cluster的对应关系，从代码上可以看到实现弄出所有的cluster，然后遍历instance集合，如果没有某个instance没有cluster，就设置成默认DEFAULT_CLUSTER_NAME，如果某个cluster没有的话就创建。然后塞到一个cluster与instance集合对应关系的map中。

接着就是遍历clusterMap更新下instance列表，这个更新instance列表代码很多，我就不贴出来了，主要思想还是比对新老的，然后找出新的instance，与挂了的instance，注意这一步是更新 cluster对象里面的集合，其实就是2个set，一个存临时节点的，一个是存永久节点的。

// 持久实例集合
@JsonIgnore
private Set<Instance> persistentInstances = new HashSet<>();

// 临时实例集合
@JsonIgnore
private Set<Instance> ephemeralInstances = new HashSet<>();

好了，到这我们的服务注册就算是完事了，最后这个service.onChange方法中还有一行代码我们需要注意 getPushService().serviceChanged(this);这个就是service服务列表发生变化，然后进行通知的。

4. 流程图

4.1 调用关系图

4.2 服务注册流程图

4.3 异步事件通知修改注册信息

4.4 图解namespace/group/service/cluster/instance的关系

代码设计上关系：

个人对namespace/group/service/cluster/instance之间关系的理解

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