nacos源码阅读--服务发现

古城微笑少年丶 2022-11-05 15:54 252阅读 0赞

**使用FeignClient的方式是怎么调用远程的微服务的，以及nacos收到请求之后是怎么做出回复的，以下来剖析：**

## 一、客户端使用FeignClient远端请求 ##

**1.1 一个最简单的请求接口：**

@Service
    @FeignClient("provider")
    public interface Feign { 
        @GetMapping("hello")
        String hello();
    }

**1.2 为了方便测试，在controller里直接调用：**

@Autowired
        Feign feign;
    
        @GetMapping("consumer")
        void operate(){ 
            String response = feign.hello();
            System.out.println(response);
        }

**1.3 通过debug测试发现发送请求的具体方法在`execute`当中（找了很久才找到），其中主要的代码也是return这一行代码，首先生成了一个client，封装了许多请求参数以及配置，然后通过`executeWithLoadBalancer`方法来进行发送请求，最后通过`toResponse`方法将返回结果进行转换，这里只看一下发送方法的最终实现**

public Response execute(Request request, Options options) throws IOException { 
            try { 
                URI asUri = URI.create(request.url());
                String clientName = asUri.getHost();
                URI uriWithoutHost = cleanUrl(request.url(), clientName);
                RibbonRequest ribbonRequest = new RibbonRequest(this.delegate, request, uriWithoutHost);
                IClientConfig requestConfig = this.getClientConfig(options, clientName);
                return ((RibbonResponse)this.lbClient(clientName).executeWithLoadBalancer(ribbonRequest, requestConfig)).toResponse();
            } catch (ClientException var8) { 
                IOException io = this.findIOException(var8);
                if (io != null) { 
                    throw io;
                } else { 
                    throw new RuntimeException(var8);
                }
            }
        }

**1.4 代码源头可以看到`convertAndSend`方法使用了jdk自带的`HttpURLConnection` 工具来进行http的请求，请求地址为`http://192.168.16.182:8889/hello`，hello是feignClient接口中的方法名。这里是服务端的真是ip以及真是端口，但是客户端是怎么知道服务端的真实ip和端口的呢？？**

HttpURLConnection convertAndSend(Request request, Options options) throws IOException { 
                URL url = new URL(request.url());
                HttpURLConnection connection = this.getConnection(url);
                if (connection instanceof HttpsURLConnection) { 
                    HttpsURLConnection sslCon = (HttpsURLConnection)connection;
                    if (this.sslContextFactory != null) { 
                        sslCon.setSSLSocketFactory(this.sslContextFactory);
                    }
    
                    if (this.hostnameVerifier != null) { 
                        sslCon.setHostnameVerifier(this.hostnameVerifier);
                    }
                }
    
                connection.setConnectTimeout(options.connectTimeoutMillis());
                connection.setReadTimeout(options.readTimeoutMillis());
                connection.setAllowUserInteraction(false);
                connection.setInstanceFollowRedirects(options.isFollowRedirects());
                connection.setRequestMethod(request.httpMethod().name());
                Collection<String> contentEncodingValues = (Collection)request.headers().get("Content-Encoding");
                boolean gzipEncodedRequest = contentEncodingValues != null && contentEncodingValues.contains("gzip");
                boolean deflateEncodedRequest = contentEncodingValues != null && contentEncodingValues.contains("deflate");
                boolean hasAcceptHeader = false;
                Integer contentLength = null;
                Iterator var10 = request.headers().keySet().iterator();
    
                while(var10.hasNext()) { 
                    String field = (String)var10.next();
                    if (field.equalsIgnoreCase("Accept")) { 
                        hasAcceptHeader = true;
                    }
    
                    Iterator var12 = ((Collection)request.headers().get(field)).iterator();
    
                    while(var12.hasNext()) { 
                        String value = (String)var12.next();
                        if (field.equals("Content-Length")) { 
                            if (!gzipEncodedRequest && !deflateEncodedRequest) { 
                                contentLength = Integer.valueOf(value);
                                connection.addRequestProperty(field, value);
                            }
                        } else { 
                            connection.addRequestProperty(field, value);
                        }
                    }
                }
    
                if (!hasAcceptHeader) { 
                    connection.addRequestProperty("Accept", "*/*");
                }
    
                if (request.requestBody().asBytes() != null) { 
                    if (contentLength != null) { 
                        connection.setFixedLengthStreamingMode(contentLength);
                    } else { 
                        connection.setChunkedStreamingMode(8196);
                    }
    
                    connection.setDoOutput(true);
                    OutputStream out = connection.getOutputStream();
                    if (gzipEncodedRequest) { 
                        out = new GZIPOutputStream((OutputStream)out); } else if (deflateEncodedRequest) { out = new DeflaterOutputStream((OutputStream)out); } try { ((OutputStream)out).write(request.requestBody().asBytes());
                    } finally { 
                        try { 
                            ((OutputStream)out).close();
                        } catch (IOException var19) { 
                        }
    
                    }
                }
    
                return connection;
            }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODk4MTkx_size_16_color_FFFFFF_t_70]

**1.5 debug有如下日志：  
字面意思是动态服务列表，进去看看源代码**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODk4MTkx_size_16_color_FFFFFF_t_70 1]  
**有一个定时任务`scheduleWithFixedDelay`，会一直周期执行更新操作：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODk4MTkx_size_16_color_FFFFFF_t_70 2]

**一直找进去，发现有一个`scheduleUpdateIfAbsent`的方法，里边又有一个`UpdateTask`的异步任务，知道最终有一个`updateServiceNow`的更新服务信息的方法：**

public void updateServiceNow(String serviceName, String clusters) { 
            ServiceInfo oldService = this.getServiceInfo0(serviceName, clusters);
            boolean var15 = false;
    
            label121: { 
                try { 
                    var15 = true;
                    String result = this.serverProxy.queryList(serviceName, clusters, this.pushReceiver.getUDPPort(), false);
                    if (StringUtils.isNotEmpty(result)) { 
                        this.processServiceJSON(result);
                        var15 = false;
                    } else { 
                        var15 = false;
                    }
                    break label121;
                } catch (Exception var19) { 
                    LogUtils.NAMING_LOGGER.error("[NA] failed to update serviceName: " + serviceName, var19);
                    var15 = false;
                } finally { 
                    if (var15) { 
                        if (oldService != null) { 
                            synchronized(oldService) { 
                                oldService.notifyAll();
                            }
                        }
    
                    }
                }
    
                if (oldService != null) { 
                    synchronized(oldService) { 
                        oldService.notifyAll();
                    }
                }
    
                return;
            }
    
            if (oldService != null) { 
                synchronized(oldService) { 
                    oldService.notifyAll();
                }
            }
    
        }

**1.6 最终发现是通过http请求来定时请求服务端，更新服务信息，接口为：`/nacos/v1/ns/instance/list`**

public String queryList(String serviceName, String clusters, int udpPort, boolean healthyOnly) throws NacosException { 
        Map<String, String> params = new HashMap(8);
        params.put("namespaceId", this.namespaceId);
        params.put("serviceName", serviceName);
        params.put("clusters", clusters);
        params.put("udpPort", String.valueOf(udpPort));
        params.put("clientIP", NetUtils.localIP());
        params.put("healthyOnly", String.valueOf(healthyOnly));
        return this.reqAPI(UtilAndComs.NACOS_URL_BASE + "/instance/list", params, (String)"GET");
    }

## 二、nacos服务端的处理 ##

**2.1 之前看过nacos注册动作源码的应该都知道最终service的信息存放在`ServiceManager`中的`consistencyService`，服务的取消注册也是通过操作`consistencyService`来进行的**

private void removeInstance(String namespaceId, String serviceName, boolean ephemeral, Service service,
            Instance... ips) throws NacosException { 
    
        String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);
    
        List<Instance> instanceList = substractIpAddresses(service, ephemeral, ips);
    
        Instances instances = new Instances();
        instances.setInstanceList(instanceList);
    
        consistencyService.put(key, instances);
    }

**`consistencyService`当中有一个监听器方法，注册以及取消注册都会触发`onChange`方法，最终同步到一个名叫`serviceMap`的变量当中以及`persistentConsistencyService`和`ephemeralConsistencyService`**

public DelegateConsistencyServiceImpl(PersistentConsistencyServiceDelegateImpl persistentConsistencyService,
            EphemeralConsistencyService ephemeralConsistencyService) { 
        this.persistentConsistencyService = persistentConsistencyService;
        this.ephemeralConsistencyService = ephemeralConsistencyService;
    }

@Override
        public void onChange(String key, Service service) throws Exception { 
            try { 
                if (service == null) { 
                    Loggers.SRV_LOG.warn("received empty push from raft, key: {}", key);
                    return;
                }
    
                if (StringUtils.isBlank(service.getNamespaceId())) { 
                    service.setNamespaceId(Constants.DEFAULT_NAMESPACE_ID);
                }
    
                Loggers.RAFT.info("[RAFT-NOTIFIER] datum is changed, key: {}, value: {}", key, service);
    
                Service oldDom = getService(service.getNamespaceId(), service.getName());
    
                if (oldDom != null) { 
                    oldDom.update(service);
                    // re-listen to handle the situation when the underlying listener is removed:
                    consistencyService
                            .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true),
                                    oldDom);
                    consistencyService
                            .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), false),
                                    oldDom);
                } else { 
                    putServiceAndInit(service);
                }
            } catch (Throwable e) { 
                Loggers.SRV_LOG.error("[NACOS-SERVICE] error while processing service update", e);
            }
        }

**2.2 知道了这些，再来找获取服务的具体操作，找到客户端请求的那个接口，在`InstanceService`类当中有一个方法名叫`list`，就是客户端获取服务请求的接口，前面做了一些参数解析操作，获取的方法为最后return方法**

/**
         * Get all instance of input service.
         *
         * @param request http request
         * @return list of instance
         * @throws Exception any error during list
         */
        @GetMapping("/list")
        @Secured(parser = NamingResourceParser.class, action = ActionTypes.READ)
        public ObjectNode list(HttpServletRequest request) throws Exception { 
            
            String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
            String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
            NamingUtils.checkServiceNameFormat(serviceName);
            
            String agent = WebUtils.getUserAgent(request);
            String clusters = WebUtils.optional(request, "clusters", StringUtils.EMPTY);
            String clientIP = WebUtils.optional(request, "clientIP", StringUtils.EMPTY);
            int udpPort = Integer.parseInt(WebUtils.optional(request, "udpPort", "0"));
            String env = WebUtils.optional(request, "env", StringUtils.EMPTY);
            boolean isCheck = Boolean.parseBoolean(WebUtils.optional(request, "isCheck", "false"));
            
            String app = WebUtils.optional(request, "app", StringUtils.EMPTY);
            
            String tenant = WebUtils.optional(request, "tid", StringUtils.EMPTY);
            
            boolean healthyOnly = Boolean.parseBoolean(WebUtils.optional(request, "healthyOnly", "false"));
            
            return doSrvIpxt(namespaceId, serviceName, agent, clusters, clientIP, udpPort, env, isCheck, app, tenant,
                    healthyOnly);
        }

**2.3 `doSrvIpxt`方法如下，可以看到又是`ServiceManager`服务管理器的`getService`方法来进行获取服务信息，这里获取到了服务信息，没有端口信息，通过debug发现，在下方的`service.srvIPs`方法解析了具体获取服务的端口**

/**
         * Get service full information with instances.
         *
         * @param namespaceId namespace id
         * @param serviceName service name
         * @param agent       agent infor string
         * @param clusters    cluster names
         * @param clientIP    client ip
         * @param udpPort     push udp port
         * @param env         env
         * @param isCheck     is check request
         * @param app         app name
         * @param tid         tenant
         * @param healthyOnly whether only for healthy check
         * @return service full information with instances
         * @throws Exception any error during handle
         */
        public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP,
                int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception { 
            
            ClientInfo clientInfo = new ClientInfo(agent);
            ObjectNode result = JacksonUtils.createEmptyJsonNode();
            Service service = serviceManager.getService(namespaceId, serviceName);
            long cacheMillis = switchDomain.getDefaultCacheMillis();
            
            // now try to enable the push
            try { 
                if (udpPort > 0 && pushService.canEnablePush(agent)) { 
                    
                    pushService
                            .addClient(namespaceId, serviceName, clusters, agent, new InetSocketAddress(clientIP, udpPort),
                                    pushDataSource, tid, app);
                    cacheMillis = switchDomain.getPushCacheMillis(serviceName);
                }
            } catch (Exception e) { 
                Loggers.SRV_LOG
                        .error("[NACOS-API] failed to added push client {}, {}:{}", clientInfo, clientIP, udpPort, e);
                cacheMillis = switchDomain.getDefaultCacheMillis();
            }
            
            if (service == null) { 
                if (Loggers.SRV_LOG.isDebugEnabled()) { 
                    Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);
                }
                result.put("name", serviceName);
                result.put("clusters", clusters);
                result.put("cacheMillis", cacheMillis);
                result.replace("hosts", JacksonUtils.createEmptyArrayNode());
                return result;
            }
            
            checkIfDisabled(service);
            
            List<Instance> srvedIPs;
            
            srvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ",")));
            
            // filter ips using selector:
            if (service.getSelector() != null && StringUtils.isNotBlank(clientIP)) { 
                srvedIPs = service.getSelector().select(clientIP, srvedIPs);
            }
            
            if (CollectionUtils.isEmpty(srvedIPs)) { 
                
                if (Loggers.SRV_LOG.isDebugEnabled()) { 
                    Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);
                }
                
                if (clientInfo.type == ClientInfo.ClientType.JAVA
                        && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) { 
                    result.put("dom", serviceName);
                } else { 
                    result.put("dom", NamingUtils.getServiceName(serviceName));
                }
                
                result.put("name", serviceName);
                result.put("cacheMillis", cacheMillis);
                result.put("lastRefTime", System.currentTimeMillis());
                result.put("checksum", service.getChecksum());
                result.put("useSpecifiedURL", false);
                result.put("clusters", clusters);
                result.put("env", env);
                result.set("hosts", JacksonUtils.createEmptyArrayNode());
                result.set("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));
                return result;
            }
            
            Map<Boolean, List<Instance>> ipMap = new HashMap<>(2);
            ipMap.put(Boolean.TRUE, new ArrayList<>());
            ipMap.put(Boolean.FALSE, new ArrayList<>());
            
            for (Instance ip : srvedIPs) { 
                ipMap.get(ip.isHealthy()).add(ip);
            }
            
            if (isCheck) { 
                result.put("reachProtectThreshold", false);
            }
            
            double threshold = service.getProtectThreshold();
            
            if ((float) ipMap.get(Boolean.TRUE).size() / srvedIPs.size() <= threshold) { 
                
                Loggers.SRV_LOG.warn("protect threshold reached, return all ips, service: {}", serviceName);
                if (isCheck) { 
                    result.put("reachProtectThreshold", true);
                }
                
                ipMap.get(Boolean.TRUE).addAll(ipMap.get(Boolean.FALSE));
                ipMap.get(Boolean.FALSE).clear();
            }
            
            if (isCheck) { 
                result.put("protectThreshold", service.getProtectThreshold());
                result.put("reachLocalSiteCallThreshold", false);
                
                return JacksonUtils.createEmptyJsonNode();
            }
            
            ArrayNode hosts = JacksonUtils.createEmptyArrayNode();
            
            for (Map.Entry<Boolean, List<Instance>> entry : ipMap.entrySet()) { 
                List<Instance> ips = entry.getValue();
                
                if (healthyOnly && !entry.getKey()) { 
                    continue;
                }
                
                for (Instance instance : ips) { 
                    
                    // remove disabled instance:
                    if (!instance.isEnabled()) { 
                        continue;
                    }
                    
                    ObjectNode ipObj = JacksonUtils.createEmptyJsonNode();
                    
                    ipObj.put("ip", instance.getIp());
                    ipObj.put("port", instance.getPort());
                    // deprecated since nacos 1.0.0:
                    ipObj.put("valid", entry.getKey());
                    ipObj.put("healthy", entry.getKey());
                    ipObj.put("marked", instance.isMarked());
                    ipObj.put("instanceId", instance.getInstanceId());
                    ipObj.set("metadata", JacksonUtils.transferToJsonNode(instance.getMetadata()));
                    ipObj.put("enabled", instance.isEnabled());
                    ipObj.put("weight", instance.getWeight());
                    ipObj.put("clusterName", instance.getClusterName());
                    if (clientInfo.type == ClientInfo.ClientType.JAVA
                            && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) { 
                        ipObj.put("serviceName", instance.getServiceName());
                    } else { 
                        ipObj.put("serviceName", NamingUtils.getServiceName(instance.getServiceName()));
                    }
                    
                    ipObj.put("ephemeral", instance.isEphemeral());
                    hosts.add(ipObj);
                    
                }
            }
            
            result.replace("hosts", hosts);
            if (clientInfo.type == ClientInfo.ClientType.JAVA
                    && clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) { 
                result.put("dom", serviceName);
            } else { 
                result.put("dom", NamingUtils.getServiceName(serviceName));
            }
            result.put("name", serviceName);
            result.put("cacheMillis", cacheMillis);
            result.put("lastRefTime", System.currentTimeMillis());
            result.put("checksum", service.getChecksum());
            result.put("useSpecifiedURL", false);
            result.put("clusters", clusters);
            result.put("env", env);
            result.replace("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));
            return result;
        }

**2.4 最终在`clusterObj.allIPs()`方法当中记录服务信息的变量查到服务具体的端口信息：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODk4MTkx_size_16_color_FFFFFF_t_70 3]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODk4MTkx_size_16_color_FFFFFF_t_70]: /images/20221023/a596a1c4284748e29b7f783a7573e6c8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODk4MTkx_size_16_color_FFFFFF_t_70 1]: /images/20221023/5d238cd0689c4f2dbf275c2d23bd64e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODk4MTkx_size_16_color_FFFFFF_t_70 2]: /images/20221023/25d3d5a0a87942b197a40088342d7e10.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM5ODk4MTkx_size_16_color_FFFFFF_t_70 3]: /images/20221023/72c69ceaa5614ea1a778117757c86c0d.png