微服务架构 | *3.5 Nacos 服务注册与发现的源码分析

前言

参考资料
《Spring Microservices in Action》
《Spring Cloud Alibaba 微服务原理与实战》
《B站 尚硅谷 SpringCloud 框架开发教程 周阳》

为方便理解与表达,这里把 Nacos 控制台和 Nacos 注册中心称为 Nacos 服务器(就是 web 界面那个),我们编写的业务服务称为 Nacso 客户端;

Nacos 客户端将自己注册进 Nacos 服务器。《1. 服务如何注册进 Nacos 注册中心》主要从 Nacos 客户端角度解释如何发送信息给 Nacos 服务器;《2. Nacos 服务器注册服务》主要从 Nacos 服务器角度解释注册原理;

《3. 客户端查询所有服务实例》将从服务消费者和提供者的角度,解释服务消费者如何获取提供者的所有实例。服务消费者和提供者都是 Nacos 的客户端;

《4. 客户端监听 Nacos 服务器以动态获取服务实例》从消费者客户端角度出发监听 Nacos 服务器,以动态获知提供者的变化;


1. 客户端注册进 Nacos 注册中心(客户端视角)

1.1 Spring Cloud 提供的规范标准

  • 服务要注册进 Spring Cloud 集成的 Nacos,首先要满足 Spring Cloud 提供的规范标准;
  • spring-cloud-common 包中有一个类 org.springframework.cloud.client.serviceregistry.ServiceRegistry,它是Spring Cloud 提供的服务注册的标准。集成到 Spring Cloud 中实现服务注册的组件,都会实现该接口;
public interface ServiceRegistry<R extends Registration>{
    void register(R registration); 
    void deregister(R registration); 
    void close(); 
    void setStatus(R registration,String status);
    <T> T getstatus(R registration);
}

1.2 Nacos 的自动配置类

  • spring-cloud-commons 包的 META-INF/spring.factories 中包含自动装配的配置信息。即约定 Spring Cloud 启动时,会将那些类自动注入到容器中:
Spring Cloud 自动装配.png
  • 其中 AutoServiceRegistrationAutoConfiguration(服务注册自动配置类) 是服务注册相关配置类,源码如下:
@Configuration
@Import({AutoServiceRegistrationConfiguration.class})
@ConditionalOnProperty(
    value = {"spring.cloud.service-registry.auto-registration.enabled"},
    matchIfMissing = true
)
public class AutoServiceRegistrationAutoConfiguration {
    //自动注册类
    @Autowired(required = false)
    private AutoServiceRegistration autoServiceRegistration;
    //自动注册类的配置文件
    @Autowired
    private AutoServiceRegistrationProperties properties;

    public AutoServiceRegistrationAutoConfiguration() {
    }

    //初始化函数
    @PostConstruct
    protected void init() {
        if (this.autoServiceRegistration == null && this.properties.isFailFast()) {
            throw new IllegalStateException("Auto Service Registration has been requested, but there is no AutoServiceRegistration bean");
        }
    }
}
  • 其中关键之处在于注入了一个 AutoServiceRegistration(服务注册器) 自动注册接口,该接口在 Spring Cloud 中有一个抽象实现类 AbstractAutoServiceRegistration(服务注册器抽象类)
  • 我们要用什么注册中心,该注册中心就要继承 AbstractAutoServiceRegistration(服务注册器抽象类) 抽象类;
  • 对于 Nacos 来说,使用 NacosAutoServiceRegistration(Nacos 服务注册器) 类继承 AbstractAutoServiceRegistration(服务注册器抽象类) 抽象类;
NacosAutoServiceRegistration 类依赖结构图.png

1.3 监听服务初始化事件 AbstractAutoServiceRegistration.bind()

  • 在上述中需要关注 ApplicationListener(事件监听器) 接口,它是一种事件监听机制,接口声明如下:
@FunctionalInterface
public interface ApplicationListener<E extends ApplicationEvent> extends EventListener {
    void onApplicationEvent(E var1);
}
  • AbstractAutoServiceRegistration(服务注册器抽象类) 使用 AbstractAutoServiceRegistration.bind() 方法实现了该接口,用来监听 WebServerInitializedEvent(服务初始化事件)
@Override
@SuppressWarnings("deprecation")
//【断点步入】
public void onApplicationEvent(WebServerInitializedEvent event) {
    bind(event);
}
  • 我们给 AbstractAutoServiceRegistration.bind() 方法打上断点,启动服务提供者,可以发现:
    • Nacos 服务器上没有服务注册实例;
    • 服务初始化已经完成;


      服务初始化完成.png
  • 【结论】说明服务注册到 Nacos 的流程是先进行服务初始化,然后通过事件监听机制监听初始化事件。当初始化完成时,调用 AbstractAutoServiceRegistration.bind() 方法将服务注册进 Nacos 注册中心;

1.4 注册服务实例的逻辑 NacosServiceRegistry.register()

这里能说明什么时候服务会将自己的信息发给 Nacos 服务器;

  • 我们追进 AbstractAutoServiceRegistration.bind() 方法,发现在 AbstractAutoServiceRegistration(服务注册器抽象类) 里调用 NacosServiceRegistry.register() 方法后, Nacos 服务器上出现服务实例;
register()方法.png
  • 我们追进 NacosServiceRegistry.register() 方法,方法的逻辑如下:
@Override
public void register(Registration registration) {
    //判断是否有服务 ID
    if (StringUtils.isEmpty(registration.getServiceId())) {
        log.warn("No service to register for nacos client...");
        return;
    }
    String serviceId = registration.getServiceId();  //服务 ID(service-provider)
    Instance instance = getNacosInstanceFromRegistration(registration);  //服务实例(里面有 ip、port 等信息)
    try {
        //【断点步入】注册的方法
        namingService.registerInstance(serviceId, instance);
        log.info("nacos registry, {} {}:{} register finished", serviceId,
                instance.getIp(), instance.getPort());
    }
    catch (Exception e) {
        log.error("nacos registry, {} register failed...{},", serviceId,
                registration.toString(), e);
    }
}
服务 ID 与服务实例信息.png
  • 我们点进去 namingService.registerInstance() 方法(实现逻辑在 NacosNamingService.registerInstance())得到注册的具体逻辑,如下:
public void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {
    if (instance.isEphemeral()) {
        //用心跳 BeatInfo 封装服务实例信息
        BeatInfo beatInfo = new BeatInfo();
        beatInfo.setServiceName(NamingUtils.getGroupedName(serviceName, groupName));
        beatInfo.setIp(instance.getIp());
        beatInfo.setPort(instance.getPort());
        beatInfo.setCluster(instance.getClusterName());
        beatInfo.setWeight(instance.getWeight());
        beatInfo.setMetadata(instance.getMetadata());
        beatInfo.setScheduled(false);
        long instanceInterval = instance.getInstanceHeartBeatInterval();
        beatInfo.setPeriod(instanceInterval == 0L ? DEFAULT_HEART_BEAT_INTERVAL : instanceInterval);
        //【断点步入 1.4.1】将 beatInfo 心跳信息放进 beatReactor 心跳发送器(发送心跳后,Nacos 服务器仍然没有服务实例)
        this.beatReactor.addBeatInfo(NamingUtils.getGroupedName(serviceName, groupName), beatInfo);
    }
    //【断点步入 1.4.2】使用 namingProxy 命名代理方式将服务实例信息以 POST 请求方式发送服务实例
    this.serverProxy.registerService(NamingUtils.getGroupedName(serviceName, groupName), groupName, instance);
}
  • 即注册服务实例的逻辑分两步:

    • 先通过 beatReactor.addBeatInfo 创建心跳信息实现健康检测;
    • 再以 POST 请求方式发送服务实例信息;

1.4.1 心跳机制 BeatReactor.addBeatInfo()

  • 我们进入 BeatReactor.addBeatInfo() 方法一探心跳机制,源码如下:
public void addBeatInfo(String serviceName, BeatInfo beatInfo) {
    LogUtils.NAMING_LOGGER.info("[BEAT] adding beat: {} to beat map.", beatInfo);
    this.dom2Beat.put(this.buildKey(serviceName, beatInfo.getIp(), beatInfo.getPort()), beatInfo);
    //【核心】定时向服务端发送一个心跳包 beatInfo
    this.executorService.schedule(new BeatReactor.BeatTask(beatInfo), 0L, TimeUnit.MILLISECONDS);
    //【核心】
    MetricsMonitor.getDom2BeatSizeMonitor().set((double)this.dom2Beat.size());
}
  • BeatReactor.addBeatInfo() 方法主要做了两件事:

    • 客户端调用 ScheduledExecutorService.schedule() 接口方法(靠 ScheduledThreadPoolExecutor(计划线程执行器) 实现)执行定时任务,在每个任务周期内定时向服务端发送一个心跳包 beatInfo;
    • 然后通过 MetricsMonitor.getDom2BeatSizeMonitor() 方法获取一个 心跳测量监视器(实际为 Gauge),不断检测服务端的回应,如果在设定时间内没有收到服务端的回应,则认为服务器出现了故障;
  • Nacos 服务端会根据客户端的心跳包不断更新服务的状态;

1.4.2 注册服务 NamingProxy.registerService()

  • 接着进入 NamingProxy.registerService() 方法,源码如下:
public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {
    LogUtils.NAMING_LOGGER.info("[REGISTER-SERVICE] {} registering service {} with instance: {}", new Object[]{this.namespaceId, serviceName, instance});
    Map<String, String> params = new HashMap(9);
    params.put("namespaceId", this.namespaceId);
    params.put("serviceName", serviceName);
    params.put("groupName", groupName);
    params.put("clusterName", instance.getClusterName());
    params.put("ip", instance.getIp());
    params.put("port", String.valueOf(instance.getPort()));
    params.put("weight", String.valueOf(instance.getWeight()));
    params.put("enable", String.valueOf(instance.isEnabled()));
    params.put("healthy", String.valueOf(instance.isHealthy()));
    params.put("ephemeral", String.valueOf(instance.isEphemeral()));
    params.put("metadata", JSON.toJSONString(instance.getMetadata()));
    //【断点步入】这步执行完后,Nacos 服务器才出现服务实例信息
    this.reqAPI(UtilAndComs.NACOS_URL_INSTANCE, params, (String)"POST");
}
  • 该方法先对服务实例做了封装,然后通过 NamingProxy.reqAPI() 方法拼凑注册服务的 API。 NamingProxy.reqAPI() 方法源码如下:
    public String reqAPI(String api, Map<String, String> params, List<String> servers, String method) {
        params.put("namespaceId", this.getNamespaceId());
        if (CollectionUtils.isEmpty(servers) && StringUtils.isEmpty(this.nacosDomain)) {
            throw new IllegalArgumentException("no server available");
        } else {
            Exception exception = new Exception();
            if (servers != null && !servers.isEmpty()) {
                Random random = new Random(System.currentTimeMillis());
                int index = random.nextInt(servers.size());

                for(int i = 0; i < servers.size(); ++i) {
                    String server = (String)servers.get(index);

                    try {
                        return this.callServer(api, params, server, method);
                    } catch (NacosException var11) {
                        exception = var11;
                        LogUtils.NAMING_LOGGER.error("request {} failed.", server, var11);
                    } catch (Exception var12) {
                        exception = var12;
                        LogUtils.NAMING_LOGGER.error("request {} failed.", server, var12);
                    }
                    index = (index + 1) % servers.size();
                }

                throw new IllegalStateException("failed to req API:" + api + " after all servers(" + servers + ") tried: " + ((Exception)exception).getMessage());
            } else {
                int i = 0;
                while(i < 3) {
                    try {
                        return this.callServer(api, params, this.nacosDomain);
                    } catch (Exception var13) {
                        exception = var13;
                        LogUtils.NAMING_LOGGER.error("[NA] req api:" + api + " failed, server(" + this.nacosDomain, var13);
                        ++i;
                    }
                }
                throw new IllegalStateException("failed to req API:/api/" + api + " after all servers(" + servers + ") tried: " + ((Exception)exception).getMessage());
            }
        }
    }

1.5 以 Open API 方式发送注册请求

  • 最终将以 Open API 方式发送如下请求给 Nacos 服务器:
  • POST 'http://127.0.0.1:8848/nacos/v1/ns/instance?servicelame=nacos.naming.serviceName&ip=10.200.9.143&port=18082'
    reqAPI() debug.png

1.6 小结

  • Nacos 在先加载配置文件初始化服务,使用 ApplicationListener(事件监听器) 监听机制监听该服务初始化事件,当服务初始化完成后,进入 NacosServiceRegistry.register() 注册逻辑;
  • 注册原理分两步:
    • 先使用 NacosNamingService(Nacos 命名服务) 发送心跳;
    • 再使用 NamingProxy(命名代理),以 POST 请求方式发送服务实例给 Nacos 服务器;
  • 即 Nacos Service 必须要确保注册的服务实例是健康的(心跳检测),才能进行服务注册;


2. Nacos 服务器注册服务(服务器视角)

  • Nacos 服务器的源码下载、启动详情请见《微服务架构 | 3.2 Alibaba Nacos 注册中心》
  • 上述提到客户端注册服务器的最后一步是向服务器发送如下 Open API 请求:
  • POST 'http://127.0.0.1:8848/nacos/v1/ns/instance?servicelame=nacos.naming.serviceName&ip=10.200.9.143&port=18082'
  • 那么服务器的源码分析将从这个请求开始;

2.1 服务器接收请求 InstanceController.register()

  • 服务器在 nacos-naming 模块下 InstanceController(服务实例控制器) 里定义了一个 register 方法用来处理客户端的注册,源码如下:
@RestController
@RequestMapping(UtilsAndCommons.NACOS_NAMING_CONTEXT + "/instance")
public class InstanceController {

    //省略其他代码

    @CanDistro
    @PostMapping
    @Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)
    public String register(HttpServletRequest request) throws Exception {
        //从请求体里解析出 namespaceId 命名空间(本例中是 public)
        final String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);
        //解析 serviceName 服务名(本例中是 DEFAULT_GROUP@@service-provider,实际就是service-provider)
        final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);
        NamingUtils.checkServiceNameFormat(serviceName);
        
        final Instance instance = parseInstance(request);
        //【断点步入】在服务器控制台注册服务实例的方法
        serviceManager.registerInstance(namespaceId, serviceName, instance);
        return "ok";
    }

        //省略其他代码
}
解析命名空间和服务名.png

2.2 在服务器控制台注册服务实例 ServiceManager.registerInstance()

  • ServiceManager.registerInstance() 方法的源码如下:
public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {
    //【断点步入 2.3】创建空服务
    createEmptyService(namespaceId, serviceName, instance.isEphemeral());
    
    Service service = getService(namespaceId, serviceName);
    
    if (service == null) {
        throw new NacosException(NacosException.INVALID_PARAM,
                "service not found, namespace: " + namespaceId + ", service: " + serviceName);
    }
    //添加服务实例
    addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);
}
  • 这段代码主要做了三个逻辑的内容:
    • createEmptyService():创建空服务(用于在 Nacos 服务器控制台的“服务列表”中展示服务信息),实际上是初始化一个 serviceMap,它是一个 ConcurrentHashMap 集合;
    • getService():从 serviceMap 中根据 namespaceld 和 serviceName 得到一个服务对象;
    • addInstance():把当前注册的服务实例保存到 Service 中;

2.3 创建空服务 ServiceManager.createEmptyService()

  • 我们一直追进去,发现最后调用的是 ServiceManager.createServiceIfAbsent() 方法,源码如下:
public void createServiceIfAbsent(String namespaceId, String serviceName, boolean local, Cluster cluster) throws NacosException {
    //通过命名空间 ID 和服务名从缓存中获取 service 服务,第一次是没有的,进入 if 语句创建 service
    Service service = getService(namespaceId, serviceName);
    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());
        service.recalculateChecksum();
        if (cluster != null) {
            cluster.setService(service);
            service.getClusterMap().put(cluster.getName(), cluster);
        }
        //判断服务是否有效,不生效将抛出异常
        service.validate();
        //【断点步入】添加与初始化服务
        putServiceAndInit(service);
        if (!local) {
            addOrReplaceService(service);
        }
    }
}
  • 第一次注册服务时缓存中肯定是没有的,我们进入到 if 语句里的 ServiceManager.putServiceAndInit() 方法;
private void putServiceAndInit(Service service) throws NacosException {
    //【断点步入 2.3.1 】将服务添加到缓存
    putService(service);
    service = getService(service.getNamespaceId(), service.getName());
    //【断点步入 2.3.2 】建立心跳机制
    service.init();
    //【断点步入 2.3.3 】实现数据一致性的监听,将服务数据进行同步
    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());
}

2.3.1 将服务添加到缓存 Service.putService()

public void putService(Service service) {
    if (!serviceMap.containsKey(service.getNamespaceId())) {
        synchronized (putServiceLock) {
            if (!serviceMap.containsKey(service.getNamespaceId())) {
                serviceMap.put(service.getNamespaceId(), new ConcurrentSkipListMap<>());
            }
        }
    }
    //将 Service 保存到 serviceMap 中
    serviceMap.get(service.getNamespaceId()).putIfAbsent(service.getName(), service);
}

2.3.2 建立心跳机制 Service.init()

public void init() {
    //【断点步入】定时检查心跳
    HealthCheckReactor.scheduleCheck(clientBeatCheckTask);
    for (Map.Entry<String, Cluster> entry : clusterMap.entrySet()) {
        entry.getValue().setService(this);
        entry.getValue().init();
    }
}
  • 进入 HealthCheckReactor.scheduleCheck() 方法;
public static void scheduleCheck(ClientBeatCheckTask task) {
    futureMap.computeIfAbsent(task.taskKey(),
            k -> GlobalExecutor.scheduleNamingHealth(task, 5000, 5000, TimeUnit.MILLISECONDS));
}
创建定时任务检查心跳.png
  • 其主要是通过创建定时任务不断检测当前服务下所有实例最后发送心跳包的时间。如果超时,则设置healthy为false表示服务不健康,并且发送服务变更事件;
  • 心跳检测机制可以用下图概述:
心跳检测机制.png

2.3.3 实现数据一致性的监听 DelegateConsistencyServiceImpl.listen()

@Override
public void listen(String key, RecordListener listener) throws NacosException {
    
    // 这个键 key 会被两个服务监听
    if (KeyBuilder.SERVICE_META_KEY_PREFIX.equals(key)) {
        //持久一致性服务
        persistentConsistencyService.listen(key, listener);
        //短暂一致性服务
        ephemeralConsistencyService.listen(key, listener);
        return;
    }
    
    mapConsistencyService(key).listen(key, listener);
}

2.4 小结

  • Nacos 客户端通过 Open API 的形式发送服务注册请求,服务端收到请求后,做以下三件事:

    • 构建一个 Service 对象保存到 ConcurrentHashMap 集合中;
    • 使用定时任务对当前服务下的所有实例建立心跳检测机制;
    • 基于数据一致性协议将服务数据进行同步;
  • 至此,基于 Nacos 客户端和服务端的服务注册原理解析可以告一段落了,下面是一些补充内容;


3. 客户端查询所有服务实例

  • 在进行 RPC 调用时,根据对服务的需求不同可以将 Nacos 客户端分为两种角色,分别是服务消费者和服务提供者;
  • 服务消费者需要使用提供者的服务,就先要向 Nacos 服务器获取所有已注册的服务提供者实例,第 3 点将讨论客户端(消费者)如何获取这些实例(提供者);

3.1 消费者客户端向 Nacos 发出请求

  • 与客户端向 Nacos 服务器注册服务一样,消费者客户端想要查询提供者的所有实例,也要向 Nacos 服务器发送请求;

  • 向 Nacos 服务器发送请求也是由两种形式,Open API 和 SDK。其中 SDK 最终也是以 Open API 方式发送的:

    • Open APIGET 127.0.0.1:8848/nacos/v1/ns/instance/list?serviceName=service-provider
    • SDKList<Instance> selectInstances(String serviceName, boolean healthy)throws NacosException;
  • 我们使用 postman 发送 GET 127.0.0.1:8848/nacos/v1/ns/instance/list?serviceName=service-provider 请求,来模拟消费者客户端获取提供者实例;

发送 postman 请求所有提供者服务实例.png
  • 这里主要多发送几次请求,因为 Nacos 服务器可能缓存了所有提供者实例信息,可能直接从缓存中拿而不是处理请求;

3.2 Nacos 服务器处理请求 InstanceController.list()

  • Nacos 服务器处理获取所有服务实例的控制器还是 nacos-naming 模块下的 InstanceController(服务实例控制器)
  • InstanceController(服务实例控制器) 提供了一个 InstanceController.list() 接口处理该请求:
Nacos 服务器处理获取服务实例的请求.png
  • 可以看到我们的断点也被捕获;
  • 源码如下:
@GetMapping("/list")
@Secured(parser = NamingResourceParser.class, action = ActionTypes.READ)
public ObjectNode list(HttpServletRequest request) throws Exception {
    
    //解析命名空间 ID
    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"));
    
    //【断点步入 3.2】获取所有服务的所有信息
    return doSrvIpxt(namespaceId, serviceName, agent, clusters, clientIP, udpPort, env, isCheck, app, tenant,
            healthyOnly);
}

3.2 获取所有服务的所有信息 InstanceController.doSrvIpxt()

  • InstanceController.doSrvIpxt() 方法很长,笔者删去一些非重点源码方便理解:
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();
    //根据 namespaceld、serviceName 获得 Service 实例
    Service service = serviceManager.getService(namespaceId, serviceName);
    List<Instance> srvedIPs;
    
    //获取指定服务下的所有实例,从Service实例中基于srvIPs得到所有服务提供者的实例信息
    srvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ",")));
    
    //使用选择器过滤一些 ip
    if (service.getSelector() != null && StringUtils.isNotBlank(clientIP)) {
        srvedIPs = service.getSelector().select(clientIP, srvedIPs);
    }
    
    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);
    }
    
    //遍历完成 JSON 字符串的封装
    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) {
            //删除不可用的实例
            if (!instance.isEnabled()) {
                continue;
            }
            ObjectNode ipObj = JacksonUtils.createEmptyJsonNode();
            ipObj.put("ip", instance.getIp());
            ipObj.put("port", instance.getPort());
            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.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;
}
  • 该方法的主要逻辑主要有三步:
    • 首先获取所有服务实例;
    • 过滤一些服务实例;
    • 遍历完成 JSON 字符串的封装并返回;


4. 客户端监听 Nacos 服务器以动态获取服务实例

  • 消费者客户端角度出发监听 Nacos 服务器,以动态获知提供者的变化;

4.1 客户端发送请求

  • 客户端服务有两种调用方式:
    • void subscribe(String serviceName, EventListener listener) throws NacosException
    • public List<Instance> selectInstances(String serviceName, List<String> clusters, boolean healthy, boolean subscribe),其中 subscribe 置为 true;

4.2 服务动态感知的原理

  • Nacos 客户端有一个 HostReactor 类,它的功能是实现服务的动态更新;
  • 客户端发起事件订阅后,在 HostReactor 中有一个 UpdateTask 线程,每 10s 发送一次 Pull 请求,获得服务端最新的地址列表;
  • 对于服务端,它和服务提供者的实例之间维持了心跳检测,一旦服务提供者出现异常,则会发送一个 Push 消息给 Nacos 客户端,也就是服务消费者;
  • 服务消费者收到请求之后,使用 HostReactor 中提供的 processServiceJSON 解析消息,并更新本地服务地址列表;
  • 原理可以用下图总结:
服务动态感知的原理.png

4.3 Nacos 服务器处理请求

  • 服务器与本篇第 3 点相同,其中服务器的逻辑在《3.2 Nacos 服务器处理请求》;


5. 补充内容

5.1 Dubbo 的自动装配

  • Spring Cloud Alibaba Dubbo 集成 Nacos 时,服务的注册是依托 Dubbo 中的自动装配机制完成的;
  • spring-cloud-alibaba-dubbo 下的 META-INF/spring.factories 文件中自动装配了一个和服务注册相关的配置类 DubboServiceRegistrationNonWebApplicationAutoConfiguration(Dubbo 注册自动配置类)
Dubbo 的自动配置文件.png
@Configuration
@ConditionalOnNotWebApplication
@ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true)
@AutoConfigureAfter(DubboServiceRegistrationAutoConfiguration.class)
@Aspect
public class DubboServiceRegistrationNonWebApplicationAutoConfiguration {

    @Autowired
    private ServiceRegistry serviceRegistry; //实现类为 NacosServiceRegistry
    ...

    //监听 ApplicationStartedEvent 事件,该事件在刷新上下文之后,调用 application 命令前触发;
    @EventListener(ApplicationStartedEvent.class)
    public void onApplicationStarted() {
        setServerPort();
        //最终调用 NacosServiceRegistry.registry 方法实现服务注册
        register();
    }
    private void register() {
        if (registered) {
            return;
        }
        //这里即 NacosServiceRegistry.registry()
        serviceRegistry.register(registration);
        registered = true;
    }
    ...
}
  • NacosServiceRegistry.registry() 的原理详情请见本篇《1.4 注册服务实例的逻辑 NacosServiceRegistry.register()》;

  • 由此可以得出结论

    • Dubbo 集成 Nacos 时,服务的注册是依赖 Dubbo 的自动装配机制;
    • 而 Dubbo 的自动装配机制是依靠 NacosServiceRegistry.register() 实现的;


6. 源码结构图总结

源码结构图大致与本篇目录一致;

6.1 客户端视角下的服务注册结构图

  • AbstractAutoServiceRegistration.bind():监听服务初始化事件;
    • NacosServiceRegistry.register():注册服务实例;
      • NacosNamingService.registerInstance():通过Nacos 命名服务注册服务实例;
        • BeatReactor.addBeatInfo():心跳机制;
          • ScheduledThreadPoolExecutor.schedule():执行定时任务,发送心跳包 beatInfo;
          • MetricsMonitor.getDom2BeatSizeMonitor():启动心跳测量监视器;
        • NamingProxy.registerService():以 Open API 方式发送注册请求;
          • NamingProxy.reqAPI():拼凑注册服务的 API;
            • NamingProxy.callServer():呼叫服务,发送请求;

6.2 服务器视角下的服务注册结构图

  • InstanceController.register():服务器接收请求;
    • ServiceManager.registerInstance():在服务器控制台注册服务实例;
      • ServiceManager.createEmptyService():创建空服务;
        • ServiceManager.createServiceIfAbsent():如果空缺就创建服务;
          • ServiceManager.putServiceAndInit():初始化服务;
            • Service.putService():将服务添加到缓存;
            • Service.init():建立心跳机制;
            • DelegateConsistencyServiceImpl.listen():实现数据一致性的监听;
      • ServiceManager.addInstance():添加服务实例;
        • KeyBuilder.buildInstanceListKey():创建服务是一致性键;
        • DelegateConsistencyServiceImpl.put():将键和服务实例放 ConsistencyService 进行一致性维护;

6.3 客户端查询所有服务实例结构图

  • InstanceController.list():服务器接收请求;
    • InstanceController.doSrvIpxt():获取所有服务的所有信息;
      • ServiceManager.getService():根据 namespaceld、serviceName 获得 service 实例;
      • Service.srvIPs():获取指定 service 服务下的所有实例;
      • JacksonUtils.createEmptyArrayNode():遍历完成 JSON 字符串的封装;



最后

\color{blue}{\rm\small{新人制作,如有错误,欢迎指出,感激不尽!}}

\color{blue}{\rm\small{欢迎关注我,并与我交流!}}

\color{blue}{\rm\small{如需转载,请标注出处!}}

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 206,013评论 6 481
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 88,205评论 2 382
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 152,370评论 0 342
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 55,168评论 1 278
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 64,153评论 5 371
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,954评论 1 283
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 38,271评论 3 399
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,916评论 0 259
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 43,382评论 1 300
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,877评论 2 323
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,989评论 1 333
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,624评论 4 322
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 39,209评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,199评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,418评论 1 260
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 45,401评论 2 352
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,700评论 2 345

推荐阅读更多精彩内容