• Nacos注册中心11-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…

    本文将解析下nacos的服务发现源码实现，nacos支持两种服务发现方式，一种是直接去nacos服务端拉取某个服务的实例列表，就像eureka那样定时去拉取注册表信息，另一种是服务订阅的方式，就是订阅某个服务，然后这个服务下面的实例列表一旦发生变化，nacos服务端就会使用udp的方式通知客户端，并将实例列表带过去。在本文中我们主要先看下 直接拉取服务实例列表的实现与订阅，对于订阅服务实例列表发生变化，nacos服务端通知订阅的客户端这部分代码我们在下篇详细介绍。

    1. 直接拉取方式

    1.1 直接拉取客户端源码

    1. Properties properties = new Properties();
    2. properties.setProperty("serverAddr", "10.0.8.247:8825");
    3. properties.setProperty("namespace","public");
    4.  
    5. NamingService naming = NamingFactory.createNamingService(properties);
    6. naming.registerInstance("userService", "11.11.11.11", 8888, "DEFAULT");
    7. List<Instance> userService = naming.getAllInstances("userService");
    8. 复制代码

    上面这段代码，先是创建一个NamingService ，接着注册了一个服务实例，最后是调用了getAllInstances方法获取某个服务的实例列表。

    我们来看下这个getAllInstances 方法实现:

    1. @Override
    2. public List<Instance> getAllInstances(String serviceName, String groupName, List<String> clusters,
    3.         boolean subscribe) throws NacosException {
    4.  
    5.     ServiceInfo serviceInfo;
    6.     // 是否订阅
    7.     if (subscribe) {
    8.         // 订阅
    9.         serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName),
    10.                 StringUtils.join(clusters, ","));
    11.     } else {
    12.         // 不进行订阅
    13.         serviceInfo = hostReactor
    14.                 .getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName),
    15.                         StringUtils.join(clusters, ","));
    16.     }
    17.     List<Instance> list;
    18.     if (serviceInfo == null || CollectionUtils.isEmpty(list = serviceInfo.getHosts())) {
    19.         return new ArrayList<Instance>();
    20.     }
    21.     return list;
    22. }
    23. 复制代码

    前面一堆重载方法，就是设置默认的group 与 cluster 我们跳过去，直接到这个参数最全的方法来，判断是否订阅，到这里默认是订阅的，也就是subscribe =true， 不过可以指定，我们就先看下这个不订阅的是怎么实现的吧，调用了getServiceInfoDirectlyFromServer 这个方法，从方法名上也能看出来，就是直接从server上获取serviceInfo:

    1. public ServiceInfo getServiceInfoDirectlyFromServer(final String serviceName, final String clusters)
    2.         throws NacosException {
    3.     String result = serverProxy.queryList(serviceName, clusters, 0, false);
    4.     if (StringUtils.isNotEmpty(result)) {
    5.         return JacksonUtils.toObj(result, ServiceInfo.class);
    6.     }
    7.     return null;
    8. }
    9. 复制代码

    这里直接是调用了serverProxy组件的queryList 方法，需要主要的是这个udp端口是0 ，因为咱们这里是不订阅的，这个upd是给订阅的接收通知用的:

    1. public String queryList(String serviceName, String clusters, int udpPort, boolean healthyOnly)
    2.         throws NacosException {
    3.  
    4.     final Map<String, String> params = new HashMap<String, String>(8);
    5.     params.put(CommonParams.NAMESPACE_ID, namespaceId);
    6.     params.put(CommonParams.SERVICE_NAME, serviceName);
    7.     params.put("clusters", clusters);
    8.     // udp端口
    9.     params.put("udpPort", String.valueOf(udpPort));
    10.     params.put("clientIP", NetUtils.localIP());
    11.     params.put("healthyOnly", String.valueOf(healthyOnly));
    12.     // 发起请求/nacos/v1/ns/instance/list GET请求
    13.     return reqApi(UtilAndComs.nacosUrlBase + "/instance/list", params, HttpMethod.GET);
    14. }
    15. 复制代码

    这个方法就是封装请求参数，udpPort，clientIP，healthyOnly 这几个参数需要注意下，接着调用reqApi 选择server 发送请求了，请求uri是 /nacos/v1/ns/instance/list ，请求方法是get。后面的选择server发送请求我们就不看了，在服务注册，服务发现里面我们看了无数次了。

    1.2 直接拉取服务端处理源码

    InstanceController 的list 方法是处理这个请求的，我们来看下:

    1. @GetMapping("/list")
    2. @Secured(parser = NamingResourceParser.class, action = ActionTypes.READ)
    3. public ObjectNode list(HttpServletRequest request) throws Exception {
    4.     // 从请求中获取各种属性
    5.     String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
    6.     String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
    7.     NamingUtils.checkServiceNameFormat(serviceName);
    8.     // agent属性用于指定提交请求的客户端是哪种类型
    9.     String agent = WebUtils.getUserAgent(request);
    10.     String clusters = WebUtils.optional(request, "clusters", StringUtils.EMPTY);
    11.     String clientIP = WebUtils.optional(request, "clientIP", StringUtils.EMPTY);
    12.     // 获取到client的端口号，后续UDP通信会使用
    13.     int udpPort = Integer.parseInt(WebUtils.optional(request, "udpPort", "0"));
    14.     String env = WebUtils.optional(request, "env", StringUtils.EMPTY);
    15.     boolean isCheck = Boolean.parseBoolean(WebUtils.optional(request, "isCheck", "false"));
    16.  
    17.     String app = WebUtils.optional(request, "app", StringUtils.EMPTY);
    18.  
    19.     String tenant = WebUtils.optional(request, "tid", StringUtils.EMPTY);
    20.  
    21.     boolean healthyOnly = Boolean.parseBoolean(WebUtils.optional(request, "healthyOnly", "false"));
    22.     // todo 对请求进行详细处理
    23.     return doSrvIpxt(namespaceId, serviceName, agent, clusters, clientIP, udpPort, env, isCheck, app, tenant,
    24.             healthyOnly);
    25. }
    26. 复制代码

    上面这一堆就是解析参数，我们不过多关注，主要看下doSrvIpxt这个方法。方法太长了我们分段看:

    1. public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP,
    2.         int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception {
    3.     // 不同agent，生成不同的clientInfo
    4.     ClientInfo clientInfo = new ClientInfo(agent);
    5.     // 创建一个JSON Node，其就是当前方法返回的结果。后续代码就是对这个Node的各种初始化
    6.     ObjectNode result = JacksonUtils.createEmptyJsonNode();
    7.     // 从注册表中获取当前服务
    8.     Service service = serviceManager.getService(namespaceId, serviceName);
    9.     long cacheMillis = switchDomain.getDefaultCacheMillis();
    10.  
    11.     // now try to enable the push
    12.     try {
    13.  
    14.         // udpPort大于0，&& 客户端语言版本判断，看能不能UDP推送
    15.         if (udpPort > 0 && pushService.canEnablePush(agent)) {
    16.             // 创建当前发出订阅请求的Nacos client的UDP Client
    17.             // 注意，在Nacos的UDP通信中，Nacos Server充当的是UDP Client，Nacos Client充当的是UDP Server
    18.             pushService
    19.                     .addClient(namespaceId, serviceName, clusters, agent, new InetSocketAddress(clientIP, udpPort),
    20.                             pushDataSource, tid, app);
    21.             cacheMillis = switchDomain.getPushCacheMillis(serviceName);
    22.         }
    23.     } catch (Exception e) {
    24.         Loggers.SRV_LOG
    25.                 .error("[NACOS-API] failed to added push client {}, {}:{}", clientInfo, clientIP, udpPort, e);
    26.         cacheMillis = switchDomain.getDefaultCacheMillis();
    27.     }
    28.     ...
    29. }
    30. 复制代码

    这一部分代码，其实就干了2件事:

    • 根据namespace与serviceName 到ServiceManager下面的serviceMap获取对应的service对象。
    • 判断upd端口与这个客户端版本啥的，看看适不适合推送，我们这里udp端口是0 ，很显然不适合。
    1. public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP,
    2.         int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception {
    3.     ...
    4.     if (service == null) {
    5.         if (Loggers.SRV_LOG.isDebugEnabled()) {
    6.             Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);
    7.         }
    8.         result.put("name", serviceName);
    9.         result.put("clusters", clusters);
    10.         result.put("cacheMillis", cacheMillis);
    11.         result.replace("hosts", JacksonUtils.createEmptyArrayNode());
    12.         return result;
    13.     }
    14.     // 代码直到这里，说明注册表中存在该服务
    15.     // 检测该服务是否被禁。若是被禁的服务，直接抛出异常
    16.     checkIfDisabled(service);
    17.  
    18.     List<Instance> srvedIPs;
    19.     // 获取到当前服务的所有实例，包含所有持久/临时实例
    20.     srvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ",")));
    21.  
    22.     // filter ips using selector:
    23.     // 若选择器不空，则根据选择算法选择可用的intance列表，默认情况下，选择器不做任务过滤
    24.     if (service.getSelector() != null && StringUtils.isNotBlank(clientIP)) {
    25.         srvedIPs = service.getSelector().select(clientIP, srvedIPs);
    26.     }
    27. }
    28. 复制代码

    首先就是service是null的时候，是null的时候组装响应参数返回给客户端了，我们看下这段代码，先是检查service 可不可用，接着就是根据cluster 获取这个service 下面的instance列表了，根据cluster获取这个服务下面对应的实例集合。

    接着就是获取这个selector ，使用selector 来过滤实例集合。

    1. public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP,
    2.         int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception {
    3.     ...
    4.     // 若最终选择的结果为空，则直接结束
    5.     if (CollectionUtils.isEmpty(srvedIPs)) {
    6.  
    7.         if (Loggers.SRV_LOG.isDebugEnabled()) {
    8.             Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);
    9.         }
    10.  
    11.         if (clientInfo.type == ClientInfo.ClientType.JAVA
    12.                 && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {
    13.             result.put("dom", serviceName);
    14.         } else {
    15.             result.put("dom", NamingUtils.getServiceName(serviceName));
    16.         }
    17.  
    18.         result.put("name", serviceName);
    19.         result.put("cacheMillis", cacheMillis);
    20.         result.put("lastRefTime", System.currentTimeMillis());
    21.         result.put("checksum", service.getChecksum());
    22.         result.put("useSpecifiedURL", false);
    23.         result.put("clusters", clusters);
    24.         result.put("env", env);
    25.         // 注意，hosts为空
    26.         result.set("hosts", JacksonUtils.createEmptyArrayNode());
    27.         result.set("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));
    28.         return result;
    29.     }
    30.     // 代码走到这里，说明具有可用的instance
    31.     Map<Boolean, List<Instance>> ipMap = new HashMap<>(2);
    32.     // 这个map只有两个key，True与False
    33.     // key为true的value中存放的是所有健康的instance
    34.     // key为false的value存放的是所有不健康的instance
    35.     ipMap.put(Boolean.TRUE, new ArrayList<>());
    36.     ipMap.put(Boolean.FALSE, new ArrayList<>());
    37.     // 根据instance的健康状态，将所有instance分流放入map的不同key的value中
    38.     for (Instance ip : srvedIPs) {
    39.         // 这个语句写的非常好
    40.         ipMap.get(ip.isHealthy()).add(ip);
    41.     }
    42.     // isCheck为true，表示需要检测instance的保护阈值
    43.     if (isCheck) {
    44.         // reachProtectThreshold 是否达到保护阈值
    45.         result.put("reachProtectThreshold", false);
    46.     }
    47.     // 获取服务的保护阈值
    48.     double threshold = service.getProtectThreshold();
    49.     // 若  "健康instance数量/instance总数" <= 保护阈值，则说明需要启动保护机制了
    50.     if ((float) ipMap.get(Boolean.TRUE).size() / srvedIPs.size() <= threshold) {
    51.  
    52.         Loggers.SRV_LOG.warn("protect threshold reached, return all ips, service: {}", serviceName);
    53.         if (isCheck) {
    54.             result.put("reachProtectThreshold", true);
    55.         }
    56.         // 将所有不健康的instance添加到的key为true的instance列表，
    57.         // 即key为true的value中（instance列表）存放的是所有instance实例
    58.         // 包含所有健康的与不健康的instance
    59.         ipMap.get(Boolean.TRUE).addAll(ipMap.get(Boolean.FALSE));
    60.         // 清空key为false的value（不健康的instance列表）
    61.         ipMap.get(Boolean.FALSE).clear();
    62.     }
    63.  
    64.     ...
    65. }
    66. 复制代码

    接着就是 创建一个map，然后用来区分健康的实例与不健康的实例，健康的实例key就是 Boolean.TRUE 不健康的就是Boolean.FALSE ，遍历区分开，接着就是判断是否检查，默认是false的，获取服务保护的阈值，默认是0 ， 如果健康的服务实例数量占比小于这个阈值的话，他就会将不健康的实例也放到健康的里面，这就是nacos的服务保护机制，可以联想下eureka的服务保护机制，为啥要把不健康的放到健康里面，因为后面返回给客户端的时候，只返回健康的这个key下面，这样就达到了活着实例数量低于某个阈值后不进行服务摘除的效果。

    后面就是遍历那个map只要Boolean.TRUE 与可用的服务，然后塞到hosts 这个map中。最后就是封装返回结果，然后返回给客户端。

    1. public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP,
    2.         int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception {
    3.     
    4.     ...
    5.     if (isCheck) {
    6.         result.put("protectThreshold", service.getProtectThreshold());
    7.         result.put("reachLocalSiteCallThreshold", false);
    8.  
    9.         return JacksonUtils.createEmptyJsonNode();
    10.     }
    11.  
    12.     ArrayNode hosts = JacksonUtils.createEmptyArrayNode();
    13.     // 注意，这个ipMap中存放着所有健康与不健康的instance列表
    14.     for (Map.Entry<Boolean, List<Instance>> entry : ipMap.entrySet()) {
    15.         List<Instance> ips = entry.getValue();
    16.         // 若客户端只要健康的instance，且当前遍历的map的key为false，则跳过
    17.         if (healthyOnly && !entry.getKey()) {
    18.             continue;
    19.         }
    20.         // 遍历的这个ips可能是所有不健康的instance列表，
    21.         // 也可能是所有健康的instance列表，
    22.         // 也可能是所有健康与不健康的instance列表总和
    23.         for (Instance instance : ips) {
    24.  
    25.             // remove disabled instance:
    26.             // 跳过禁用的instance
    27.             if (!instance.isEnabled()) {
    28.                 continue;
    29.             }
    30.  
    31.             ObjectNode ipObj = JacksonUtils.createEmptyJsonNode();
    32.             // 将当前遍历的instance转换为JSON
    33.             ipObj.put("ip", instance.getIp());
    34.             ipObj.put("port", instance.getPort());
    35.             // deprecated since nacos 1.0.0:
    36.             ipObj.put("valid", entry.getKey());
    37.             ipObj.put("healthy", entry.getKey());
    38.             ipObj.put("marked", instance.isMarked());
    39.             ipObj.put("instanceId", instance.getInstanceId());
    40.             ipObj.set("metadata", JacksonUtils.transferToJsonNode(instance.getMetadata()));
    41.             ipObj.put("enabled", instance.isEnabled());
    42.             ipObj.put("weight", instance.getWeight());
    43.             ipObj.put("clusterName", instance.getClusterName());
    44.             if (clientInfo.type == ClientInfo.ClientType.JAVA
    45.                     && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {
    46.                 ipObj.put("serviceName", instance.getServiceName());
    47.             } else {
    48.                 ipObj.put("serviceName", NamingUtils.getServiceName(instance.getServiceName()));
    49.             }
    50.  
    51.             ipObj.put("ephemeral", instance.isEphemeral());
    52.             hosts.add(ipObj);
    53.  
    54.         }
    55.     }
    56.  
    57.     result.replace("hosts", hosts);
    58.     if (clientInfo.type == ClientInfo.ClientType.JAVA
    59.             && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {
    60.         result.put("dom", serviceName);
    61.     } else {
    62.         result.put("dom", NamingUtils.getServiceName(serviceName));
    63.     }
    64.     result.put("name", serviceName);
    65.     result.put("cacheMillis", cacheMillis);
    66.     result.put("lastRefTime", System.currentTimeMillis());
    67.     result.put("checksum", service.getChecksum());
    68.     result.put("useSpecifiedURL", false);
    69.     result.put("clusters", clusters);
    70.     result.put("env", env);
    71.     result.replace("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));
    72.     return result;
    73. }
    74. 复制代码

    好了我们直接去服务端来去服务是列表的源码就看完了。

    2. 订阅通知方式

    2.1 客户端发起订阅

    1. serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName),
    2.                     StringUtils.join(clusters, ","));
    3. 复制代码

    接着1.2小节getAllInstances 这个方法分析，这次我们看下订阅是怎么弄的:

    2.1.1 HostReactor#getServiceInfo

    1. public ServiceInfo getServiceInfo(final String serviceName, final String clusters) {
    2.  
    3.     NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch());
    4.     // key的格式为：groupId@@微服务名称@@clusters名称
    5.     String key = ServiceInfo.getKey(serviceName, clusters);
    6.     if (failoverReactor.isFailoverSwitch()) {
    7.         return failoverReactor.getService(key);
    8.     }
    9.  
    10.     // todo 从当前Client本地注册表中获取当前服务
    11.     ServiceInfo serviceObj = getServiceInfo0(serviceName, clusters);
    12.  
    13.     // 若本地注册表中没有该服务，则创建一个
    14.     if (null == serviceObj) {
    15.         // 创建一个空的服务（没有任何提供者实例instance的ServiceInfo）
    16.         serviceObj = new ServiceInfo(serviceName, clusters);
    17.  
    18.         serviceInfoMap.put(serviceObj.getKey(), serviceObj);
    19.  
    20.         // updatingMap 是一个临时缓存，主要使用这个map的key
    21.         // map的key不能重复的特性
    22.         // 只要这个服务名称在这个map中，说明这个服务正在更新中
    23.         updatingMap.put(serviceName, new Object());
    24.         // todo 更新本地注册表ServiceName的服务
    25.         updateServiceNow(serviceName, clusters);
    26.         // 更新完毕，从updatingMap中删除
    27.         updatingMap.remove(serviceName);
    28.  
    29.     // 若当前注册表中已经有这个服务，那么查看一下临时map下
    30.     // 是否存在该服务，若存在，说明当前服务正在更新中，所以本次操作先等待一段时间，默认5s
    31.     } else if (updatingMap.containsKey(serviceName)) {
    32.  
    33.         if (UPDATE_HOLD_INTERVAL > 0) {
    34.             // hold a moment waiting for update finish
    35.             synchronized (serviceObj) {
    36.                 try {
    37.                     serviceObj.wait(UPDATE_HOLD_INTERVAL);
    38.                 } catch (InterruptedException e) {
    39.                     NAMING_LOGGER
    40.                             .error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e);
    41.                 }
    42.             }
    43.         }
    44.     }
    45.  
    46.     // todo 启动一个定时任务，定时更新本地注册表中的当前服务
    47.     scheduleUpdateIfAbsent(serviceName, clusters);
    48.  
    49.     return serviceInfoMap.get(serviceObj.getKey());
    50. }
    51. 复制代码

    先是根据clusters与serviceName生成一个订阅key，接着就是调用getServiceInfo0 方法获取本地的一个缓存:

    1. private ServiceInfo getServiceInfo0(String serviceName, String clusters) {
    2.  
    3.     String key = ServiceInfo.getKey(serviceName, clusters);
    4.  
    5.     // serviceInfoMap 是client的本地注册表
    6.     // key:groupId@@微服务名称@@clusters名称 value:ServiceInfo
    7.     return serviceInfoMap.get(key);
    8. }
    9. 复制代码

    其实就是与上面生成那个key一样，然后去serviceInfoMap 这个map中获取，它你可以理解成一个本地的缓存。 如果没有的话，就创建一个ServiceInfo 对象，放到serviceInfoMap 这个map中，然后在updatingMap 中插入这serviceName ，表示这个serviceName正在更新。调用updateServiceNow这个方法立即更新，这个其实就是立即发送请求去客户端端获取。我们看下

    2.1.2 HostReactor#updateServiceNow

    1. private void updateServiceNow(String serviceName, String clusters) {
    2.     try {
    3.         // todo
    4.         updateService(serviceName, clusters);
    5.     } catch (NacosException e) {
    6.         NAMING_LOGGER.error("[NA] failed to update serviceName: " + serviceName, e);
    7.     }
    8. }
    9.  
    10. public void updateService(String serviceName, String clusters) throws NacosException {
    11.     // 本地注册表中获取当前服务
    12.     ServiceInfo oldService = getServiceInfo0(serviceName, clusters);
    13.     try {
    14.         // todo 提交get请求，获取服务ServiceInfo
    15.         // 需要注意，返回的是json串
    16.         String result = serverProxy.queryList(serviceName, clusters, pushReceiver.getUdpPort(), false);
    17.  
    18.         if (StringUtils.isNotEmpty(result)) {
    19.             // todo 将来自Server的ServiceInfo更新到本地
    20.             processServiceJson(result);
    21.         }
    22.     } finally {
    23.         if (oldService != null) {
    24.             synchronized (oldService) {
    25.                 oldService.notifyAll();
    26.             }
    27.         }
    28.     }
    29. }
    30. 复制代码

    这里它又从缓存中获取了一次老的，接着就是调用serverProxy 这个组件的queryList 方法来获取服务实例列表了，这个方法很熟悉，我们第1.2小节见过它，需要注意的是pushReceiver.getUdpPort() ，就是获取了一个upd端口，这个pushReceiver 组件就是用来接收nacos服务端推送的，后面我们在看下，关于serverProxy.queryList 往后的代码我们就不看了。这里推荐看 第2.2小节 服务端处理订阅 章节内容。看完之后再回来。

    回到HostReactor#getServiceInfo 这个方法中，我们上面介绍完了ServiceInfo 是null的情况，接着就是介绍updatingMap里面有serviceName，就是说明这个正在更新着，不要着急，等待5s在获取，如果如果提前有了数据，会有线程通知然后唤醒这个等待的线程。接着就是调用了 scheduleUpdateIfAbsent(serviceName, clusters);这个方法：

    1. public void scheduleUpdateIfAbsent(String serviceName, String clusters) {
    2.     // futureMap是一个缓存map，其key为 groupId@@微服务名称@@clusters
    3.     // value是一个定时异步操作对象
    4.     // 这种结构称之为：双重检测锁，DCL，Double Check Lock
    5.     // 该结构是为了避免在并发情况下，多线程重复写入数据
    6.     // 该结构的特征：
    7.     // 1)有两个不为null的判断
    8.     // 2)有共享集合
    9.     // 3)有synchronized代码块
    10.     if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
    11.         return;
    12.     }
    13.  
    14.     synchronized (futureMap) {
    15.         if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) {
    16.             return;
    17.         }
    18.         // 创建一个定时异步操作对象，并启动这个定时任务
    19.         ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, clusters));
    20.         // 将这个定时异步操作对象写入到缓存map
    21.         futureMap.put(ServiceInfo.getKey(serviceName, clusters), future);
    22.     }
    23. }
    24. 复制代码

    先是从futureMap中获取这个key ，如果有了的话直接返回，很显然我们第一次是没有的， 就会接着往下走，调用了addTask方法添加一个task，然后返回一个future，将这个future 缓存到map中去了。

    1. public synchronized ScheduledFuture<?> addTask(UpdateTask task) {
    2.     // 默认延迟1s执行
    3.     return executor.schedule(task, DEFAULT_DELAY, TimeUnit.MILLISECONDS);
    4. }
    5. 复制代码

    将task扔到一个任务调度线程池里面了，然后延迟1s调度。我们看下这个任务干啥事情

    2.1.3 UpdateTask#run

    这个方法比较长，我们一部分一部分看下。

    1. public class UpdateTask implements Runnable {
    2.     ...
    3.     @Override
    4.     public void run() {
    5.         long delayTime = DEFAULT_DELAY;
    6.  
    7.         try {
    8.             // 从本地注册表中获取当前服务
    9.             ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
    10.             // 若本地注册表中不存在该服务，则从server获取到后，更新到本地注册表
    11.             if (serviceObj == null) {
    12.                 // 从server获取当前服务，并更新到本地注册表
    13.                 updateService(serviceName, clusters);
    14.                 return;
    15.             }
    16.  
    17.             // 处理本地注册表中存在当前服务的情况
    18.             // 1)serviceObj.getLastRefTime() 获取到的是当前服务最后被访问的时间，这个时间
    19.             // 是来自于本地注册表的，其记录的是所有提供这个服务的instance中最后一个instance
    20.             // 被访问的时间
    21.             // 2)缓存lastRefTime 记录的是当前instance最后被访问的时间
    22.             // 若1)时间 小于 2)时间，说明当前注册表应该更新的
    23.             if (serviceObj.getLastRefTime() <= lastRefTime) {
    24.                 updateService(serviceName, clusters);
    25.                 serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters));
    26.             } else {
    27.                 // if serviceName already updated by push, we should not override it
    28.                 // since the push data may be different from pull through force push
    29.                 refreshOnly(serviceName, clusters);
    30.             }
    31.             ...
    32.     }
    33. }
    34. 复制代码

    这个其实就是从serviceInfoMap这个本地缓存中获取咱们订阅的服务信息ServiceInfo，如果是null的话就走updateService 方法，这个方法就是2.1.2小节介绍的了，直接去服务端拉取。 如果缓存中ServiceInfo 的最后更新时间小于这个任务里面维护的最后时间值的话，说明serviceInfoMap 缓存中的服务信息有点旧了，也就走updateService 方法去拉取，否则的话就是调用refreshOnly方法，这个方法就是告诉服务端要订阅哪个服务。

    1. public class UpdateTask implements Runnable {
    2.     ...
    3.     @Override
    4.     public void run() {
    5.     ...
    6.            
    7.         // 将来自于注册表的这个最后访问时间更新到当前client的缓存
    8.         lastRefTime = serviceObj.getLastRefTime();
    9.  
    10.         // 如果这个事件分发器没有订阅这个service 并且futureMap中也没有这个任务
    11.         // 就终止这个任务
    12.         if (!notifier.isSubscribed(serviceName, clusters) && !futureMap
    13.                 .containsKey(ServiceInfo.getKey(serviceName, clusters))) {
    14.             // abort the update task
    15.             NAMING_LOGGER.info("update task is stopped, service:" + serviceName + ", clusters:" + clusters);
    16.             return;
    17.         }
    18.         // 如果host为空就增加失败次数
    19.         if (CollectionUtils.isEmpty(serviceObj.getHosts())) {
    20.             incFailCount();
    21.             return;
    22.         }
    23.         // 这个值是由nacos服务端决定的，如果订阅了的话就是10s
    24.         delayTime = serviceObj.getCacheMillis();
    25.         // 重置失败次数
    26.         resetFailCount();
    27.     } catch (Throwable e) {
    28.         // 增加失败次数
    29.         incFailCount();
    30.         NAMING_LOGGER.warn("[NA] failed to update serviceName: " + serviceName, e);
    31.     } finally {
    32.         // 开启下一次的定时任务
    33.         executor.schedule(this, Math.min(delayTime << failCount, DEFAULT_DELAY * 60), TimeUnit.MILLISECONDS);
    34.     }
    35. }
    36. 复制代码

    这里先是更新一下任务里面维护的这个lastRefTime时间值，接着就是判断如果eventDispatcher 组件没有订阅这个服务并且 futureMap（任务集合）里面没有这个的话，就说明被任务被停了，这个任务直接return就可以了， 接着就是计算下延迟时间，然后放到调度线程池中执行，普通情况延迟10s，失败的话就多延迟会，但是不会超过60s。

    2.2 服务端处理订阅

    服务端处理订阅还是在InstanceController 的list 方法进行的，因为是一个接口，所以处理逻辑都是一样的，我们这里就不关注了，主要看下判断upd端口那段代码:

    1. // now try to enable the push
    2. try {
    3.     // udpPort大于0，&& 客户端语言版本判断，看能不能UDP推送
    4.     if (udpPort > 0 && pushService.canEnablePush(agent)) {
    5.         // 创建当前发出订阅请求的Nacos client的UDP Client
    6.         // 注意，在Nacos的UDP通信中，Nacos Server充当的是UDP Client，Nacos Client充当的是UDP Server
    7.         pushService
    8.                 .addClient(namespaceId, serviceName, clusters, agent, new InetSocketAddress(clientIP, udpPort),
    9.                         pushDataSource, tid, app);
    10.         cacheMillis = switchDomain.getPushCacheMillis(serviceName);
    11.     }
    12. } catch (Exception e) {
    13.     Loggers.SRV_LOG
    14.             .error("[NACOS-API] failed to added push client {}, {}:{}", clientInfo, clientIP, udpPort, e);
    15.     cacheMillis = switchDomain.getDefaultCacheMillis();
    16. }
    17. 复制代码

    调用了pushService 组件的addClient 方法，这个pushService 组件主要就是用来进行推送的， 比如我们订阅了某个服务，然后这个服务下面的实例信息发生了变化，pushService组件就会通知所有的订阅客户端，将新的数据给客户端推过去。

    1. public void addClient(String namespaceId, String serviceName, String clusters, String agent,
    2.         InetSocketAddress socketAddr, DataSource dataSource, String tenant, String app) {
    3.  
    4.     // 创建一个UDP Client
    5.     PushClient client = new PushClient(namespaceId, serviceName, clusters, agent, socketAddr, dataSource, tenant,
    6.             app);
    7.     // todo 将这个UDP Client添加到缓存map
    8.     addClient(client);
    9. }
    10. 复制代码

    我们看下这个 pushService组件的addClient方法，首先是创建一个PushClient对象，就是把我们订阅的一些信息封装到了PushClient这个对象中。接着就是调用addClient的重载方法。

    1. public void addClient(PushClient client) {
    2.     // client is stored by key 'serviceName' because notify event is driven by serviceName change
    3.     String serviceKey = UtilsAndCommons.assembleFullServiceName(client.getNamespaceId(), client.getServiceName());
    4.     // clientMap是一个缓存map，用于存放当前Nacos Server中所有instance对应的UDP Client
    5.     // 其是一个双层map，外层map的key为  namespaceId##groupId@@微服务名称，value为内层map
    6.     // 内层map的key为代表一个instance的字符串，value为该instance对应的UDP Client，即PushClient
    7.     ConcurrentMap<String, PushClient> clients = clientMap.get(serviceKey);
    8.     // 若当前服务的内层map为null，则创建一个并放入到缓存map
    9.     if (clients == null) {
    10.         clientMap.putIfAbsent(serviceKey, new ConcurrentHashMap<>(1024));
    11.         clients = clientMap.get(serviceKey);
    12.     }
    13.  
    14.     PushClient oldClient = clients.get(client.toString());
    15.     // 从内层map中获取当前instance对应的的PushClient，
    16.     // 若该PushClient不为null，则更新一个最后引用时间戳;
    17.     // 若该PushClient为null，则将当前这个PushClient作为PushClient
    18.     // 写入到内层map，即写入到了缓存map
    19.     if (oldClient != null) {
    20.         // 更新最后引用时间戳
    21.         oldClient.refresh();
    22.     } else {
    23.         PushClient res = clients.putIfAbsent(client.toString(), client);
    24.         if (res != null) {
    25.             Loggers.PUSH.warn("client: {} already associated with key {}", res.getAddrStr(), res.toString());
    26.         }
    27.         Loggers.PUSH.debug("client: {} added for serviceName: {}", client.getAddrStr(), client.getServiceName());
    28.     }
    29. }
    30. 复制代码

    这个方法就是先生成一个serviceKey ，然后去clientMap中获取，如果没有这个clients话就创建塞到这个clientMap 中， 最后就是将PushClient 转成字符串当作key，去clients这个map中获取，如果没有，就添加进去，如果有的话就只是刷新一下旧的。好了，到这就把订阅信息添加到PushService这个组件里面了，这个时候服务端就可以返回给客户端了。

    该处理方式主要完成了两项重要任务：

    • 创建了该Nacos Client对应的UDP通信客户端PushClient，并将其写入到了一个缓存map
    • 从注册表中获取到指定服务的所有可用的instance，并将其封装为JSON

    3. 总结

    总结一下，其实不订阅的服务发现非常简单，就是客户端发起请求直接去服务端拉取订阅的服务实例列表就可以了，如果是订阅的话，客户端先去服务端拉取订阅服务列表，并且客户端带上自己本地的udp端口给服务端，服务端会根据udp与客户端版本判断能不能订阅，能的话服务端根据订阅信息生成一个PushClient放到一个clientMap中，到时候服务实例信息改变的时候，就会从这个clientMap 取到这个PushClient信息，然后进行推送。客户端向服务端拉取完服务列表后，接着生成一个更新任务，隔一段时间去服务端刷新下数据啥的。

  • 相关阅读:
    基于机器视觉的移动消防机器人（三）--软件设计
    分享一个基于asp.net的供销社农产品商品销售系统的设计与实现（源码调试 lw开题报告ppt）
    JAVASE（复习）——方法重载、数组篇
    Mac M1通过VMWare Fusion安装Centos7记录(镜像和网络有大坑)
    文件的打开方式
    【办公自动化】用Python批量从上市公司年报中获取主要业务信息
    7个新的ES2022 JavaScript 功能，你千万不要错过了
    什么是分布式锁？他解决了什么样的问题？
    字符串 实现 strStr()
    react useMemo 用法
  • 原文地址：https://blog.csdn.net/BASK2311/article/details/127765549