Spring Cloud 整合

前言

玩SpringCloud之前最好懂SpringBoot，别搞撑死骆驼的事。Servlet整一下变成Spring；SSM封装、加入东西就变为SpringBoot；SpringBoot再封装、加入东西就变为SpringCloud

架构的演进单体应用架构

单体架构：表示层、业务逻辑层和数据访问层即所有功能都在一个工程里，打成一个jar包、war包进行部署，例如：GitHub 是基于 Ruby on Rails 的单体架构，直到 2021 年，为了让超过一半的开发人员在单体代码库之外富有成效地开展工作，GitHub 以赋能为出发点开始了向微服务架构的迁移

下图服务器用Tomcat举例

优点：

1. 单体架构开发简单，容易上手，开发人员只要集中精力开发当前工程
2. 容易修改，只需要修改对应功能模块的代码，且容易找到相关联的其他业务代码
3. 部署简单，由于是完整的结构体，编译打包成jar包或者war包，直接部署在一个服务器上即可
4. 容易扩展，可以将某些业务抽出一个新的单体架构，用于独立分担压力，也可以方便部署集群
5. 性能最高，对于单台服务器而言，单体架构独享内存和cpu，不需要api远程调用，性能损耗最小

缺点：

1. 灵活度不高，随着代码量增加，代码整体编译效率下降
2. 规模化，无法满足团队规模化开发，因为共同修改一个项目
3. 应用扩展性比较差，只能横向扩展，不能深度扩展，扩容只能只对这个应用进行扩容，不能做到对某个功能点进行扩容，关键性的代码改动一处多处会受影响
4. 健壮性不高，任何一个模块的错误均可能造成整个系统的宕机
5. 技术升级，如果想对技术更新换代，代价很大

演进：增加本地缓存和分布式缓存

缓存能够将经常访问的页面或信息存起来，从而不让其去直接访问数据库，从而增大数据库压力，但是：这就会把压力变成单机Tomcat来承受了，因此缺点就是：此时单机的tomcat又不足以支撑起高并发的请求

垂直应用架构：引入Nginx

搭配N个tomcat，从而对请求”均衡处理”，如：如果Nginx可以处理10000条请求，假设一个 tomcat可以处理100个请求，那么：就需要100个tomcat从而实现每个tomcat处理100个请求(假设每个tomcat的性能都一样 )

缺点就是数据库不足以支撑压力

后面就是将数据库做读写分离

后面还有数据库大表拆小表、大业务拆为小业务、复用功能抽离…………..

面向服务架构：SOA

SOA指的是Service-OrientedArchitecture，即面向服务架构

随着业务越来越多，代码越来越多，按照业务功能将本来一整块的系统拆分为各个不同的子系统分别提供不同的服务，服务之间会彼此调用，错综复杂

而SOA的思想就是基于前面拆成不同的服务之后，继续再抽离一层，搞一个和事佬，即下图的“统一接口”

这样不同服务之间调用就可以通过统一接口进行调用了，如：用户服务需要调用订单服务，那么用户服务去找统一接口，然后由统一接口去调用订单服务，从而将订单服务中需要的结果通过统一接口的http+json或其他两种格式返回给用户服务，这样订单服务就是服务提供者，用户服务就是服务消费者，而统一接口就相当于是服务的注册与发现

• 注意：上面这段话很重要，和后面要玩的微服务框架SpringCloud技术栈有关

学过设计模式的话，上面这种不就类似行为型设计模式的“中介者模式”吗

上面这种若是反应不过来，那拆回单体架构就懂了

微服务架构

微服务架构是分布式架构的具体实现方式，和Spring的IOC控制反转和DI依赖注入的关系一样，一种是理论，一种是具体实现方案

微服务架构和前面的SOA架构是孪生兄弟，即：微服务架构是在SOA架构的基础上，通过前人不断实践、不断踩坑、不断总结，添加了一些东西之后(如：链路追踪、配置管理、负债均衡…………)，从而变出来的一种经过良好架构设计的分布式架构方案

而广泛应用的方案框架之一就是 SpringCloud

其中常见的组件包括：

另外，SpringCloud底层是依赖于SpringBoot的，并且有版本的兼容关系，如下：

因此。现在系统架构就变成了下面这样，当然不是一定是下面这样架构设计，还得看看架构师，看领导

因此，微服务技术知识如下

Eureka注册中心

SpringCloud中文官网：https://www.springcloud.cc/spring-cloud-greenwich.html#netflix-ribbon-starter

SpringCloud英文网：https://spring.io/projects/spring-cloud

Eureka是什么？

Eureka是Netflix开发的服务发现框架，本身是一个基于REST的服务，主要用于定位运行在AWS域中的中间层服务，以达到负载均衡和中间层服务故障转移的目的。

SpringCloud将它集成在其子项目spring-cloud-netflix中，以实现SpringCloud的服务发现功能

偷张图更直观地了解一下：

如上图所示，服务提供方会将自己注册到EurekaServer中，这样EurekaServer就会存储各种服务信息，而服务消费方想要调用服务提供方的服务时，直接找EurekaServer拉取服务列表，然后根据特定地算法(轮询、随机……)，选择一个服务从而进行远程调用

• 服务提供方：一次业务中，被其它微服务调用的服务。（提供接口给其它微服务）
• 服务消费方：一次业务中，调用其它微服务的服务。（调用其它微服务提供的接口）

服务提供者与服务消费者的角色并不是绝对的，而是相对于业务而言

如果服务A调用了服务B，而服务B又调用了服务C，服务B的角色是什么？

• 对于A调用B的业务而言：A是服务消费者，B是服务提供者
• 对于B调用C的业务而言：B是服务消费者，C是服务提供者

因此，服务B既可以是服务提供者，也可以是服务消费者

Eureka的自我保护机制

这张图中EurekaServer和服务提供方有一个心跳检测机制，这是EurekaServer为了确定这些服务是否还在正常工作，所以进行的心跳检测

eureka-client启动时， 会开启一个心跳任务，向Eureka Server发送心跳，默认周期为30秒/次，如果Eureka Server在多个心跳周期内没有接收到某个节点的心跳，Eureka Server将会从服务注册表中把这个服务节点移除(默认90秒)

eureka-server维护了每个实例的最后一次心跳时间，客户端发送心跳包过来后，会更新这个心跳时间

eureka-server启动时，开启了一个定时任务，该任务每60s/次，检查每个实例的最后一次心跳时间是否超过90s，如果超过则认为过期，需要剔除

但是EurekaClient也会因为网络等原因导致没有及时向EurekaServer发送心跳，因此EurekaServer为了保证误删服务就会有一个“自我保护机制”，俗称“好死不如赖活着”

如果在短时间内EurekaServer丢失过多客户端时 (可能断网了，低于85%的客户端节点都没有正常的心跳 )，那么Eureka Server就认为客户端与注册中心出现了网络故障，Eureka Server自动进入自我保护状态 。Eureka的这样设计更加精准地控制是网络通信延迟，而不是服务挂掉了，一旦进入自我保护模式，那么 EurekaServer就会保留这个节点的属性，不会删除，直到这个节点恢复正常心跳

• 85% 这个阈值，可以通过如下配置来设置：

eureka:  server:    renewal-percent-threshold: 0.85

这里存在一个问题，这个85%是超过谁呢？这里有一个预期的续约数量，计算公式如下:

自我保护阀值 = 服务总数 * 每分钟续约数(60S/客户端续约间隔) * 自我保护续约百分比阀值因子

在自我保护模式中，EurekaServer会保留注册表中的信息，不再注销任何服务信息，当它收到正常心跳时，才会退出自我保护模式，也就是：宁可保留错误的服务注册信息，也不会盲目注销任何可能健康的服务实例，即：好死不如赖活着

因此Eureka进入自我保护状态后，会出现以下几种情况:

• Eureka Server仍然能够接受新服务的注册和查询请求，但是不会被同步到其他节点上，保证当前节点依然可用。Eureka的自我保护机制可以通过如下的方式开启或关闭

eureka:  server:#   开启Eureka自我保护机制，默认为true    enable-self-preservation: true

• Eureka Server不再从注册列表中移除因为长时间没有收到心跳而应该剔除的过期服务，如果在保护期内这个服务提供者刚好非正常下线了，此时服务消费者就会拿到一个无效的服务实例，此时会调用失败，对于这个问题需要服务消费者端要有一些容错机制，如重试，断路器等！

Eureka常用配置

eureka:  client: # eureka客户端配置    register-with-eureka: true # 是否将自己注册到eureka服务端上去    fetch-registry: true # 是否获取eureka服务端上注册的服务列表    service-url:      defaultZone: http://localhost:8001/eureka/ # 指定注册中心地址。若是集群可以写多个，中间用 逗号 隔开    enabled: true # 启用eureka客户端    registry-fetch-interval-seconds: 30 # 定义去eureka服务端获取服务列表的时间间隔  instance: # eureka客户端实例配置    lease-renewal-interval-in-seconds: 30 # 定义服务多久去注册中心续约    lease-expiration-duration-in-seconds: 90 # 定义服务多久不去续约认为服务失效    metadata-map:      zone: hangzhou # 所在区域    hostname: localhost # 服务主机名称    prefer-ip-address: false # 是否优先使用ip来作为主机名  server: # eureka服务端配置    enable-self-preservation: false #关 闭eureka服务端的自我保护机制

使用Eureka

实现如下的逻辑：

搭建Eureka Server

自行单独创建一个Maven项目，导入依赖如下：

  org.springframework.cloud  spring-cloud-starter-netflix-eureka-server

在YAML文件中一般可配置内容如下：

server:  port: 10086spring:  application:    name: EUREKA-SERVEReureka:  instance:    # Eureka的主机名，是为了eureka集群服务器之间好区分    hostname: 127.0.0.1    # 最后一次心跳后，间隔多久认定微服务不可用，默认90    lease-expiration-duration-in-seconds: 90  client:    # 不向注册中心注册自己。应用为单个注册中心设置为false，代表不向注册中心注册自己，默认true注册中心不需要开启    # registerWithEureka: false    # 不从注册中心拉取自身注册信息。单个注册中心则不拉取自身信息，默认true注册中心不需要开启    # fetchRegistry: false    service-url:      # Eureka Server的地址      defaultZone: http://${eureka.instance.hostname}:${server.port}/eureka#    server:#      # 开启Eureka自我保护机制，默认为true#      enable-self-preservation: true

• 注：在SpringCloud中配置文件YAML有两种方式，一种是 application.yml 另一种是 bootstrap.yml ，这个知识后续Nacos注册中心会用到，区别去这里：https://www.cnblogs.com/sharpest/p/13678443.html

启动类编写内容如下：

import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.cloud.netflix.eureka.server.EnableEurekaServer;/** * 
@description  : 该类功能  eureka server启动类 * 
 * 
@author       : ZiXieqing
 *//*@EnableEurekaServer 开启Eureka Server功能*/@EnableEurekaServer@SpringBootApplicationpublic class EurekaApplication {    public static void main(String[] args) {        SpringApplication.run(EurekaApplication.class, args);    }}

服务提供者

新建一个Maven模块项目，依赖如下：

  org.springframework.boot  spring-boot-starter-web  org.springframework.cloud  spring-cloud-starter-netflix-eureka-client

YAML配置内容如下：

server:  port: 8081spring:  application:    name: USER-SERVICEeureka:  client:    service-url:      # 将服务注册到哪个eureka server      defaultZone: http://localhost:10086/eureka

启动类内容如下：

import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic class UserApplication {    public static void main(String[] args) {        SpringApplication.run(UserApplication.class, args);    }}

关于开启Eureka Client的问题

上一节中启动类里面有些人会看到是如下的方式：

import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.cloud.netflix.eureka.EnableEurekaClient;@SpringBootApplication@EnableEurekaClient // 多了这么一个操作：开启eureka client功能public class UserApplication {    public static void main(String[] args) {        SpringApplication.run(UserApplication.class, args);    }}

在eureka client启动类中，为什么有些人会加 @EnableEurekaClient 注解，而有些人不会加上，为什么？

要弄这个问题，首先看yml中的配置，有些是在yml中做了一个操作：

eureka:  client:    service-url:      # 向哪个eureka server进行服务注册      defaultZone: http://localhost:10086/eureka    # 开启eureka client功能，默认就是true，差不多等价于启动类中加 @EnableEurekaClient 注解    enabled: true

既然上面配置默认值都是true，那还有必要在启动类中加入 @EnableEurekaClient 注解吗？

答案是根本不用加，加了也是多此一举(前提：yml配置中没有手动地把值改为false)，具体原因看源码：答案就在Eureka client对应的自动配置类 EurekaClientAutoConfiguration 中

上图中这一行的意思是只有当application.yaml（或者环境变量，或者系统变量）里，eureka.client.enabled这个属性的值为true才会初始化这个类（如果手动赋值为false，就不会初始化这个类了）

另外再加上另一个原因，同样在 EurekaClientAutoConfiguration 类中还有一个 eurekaAutoServiceRegistration() 方法

在这里使用 EurekaAutoServiceRegistration类+@Bean注解 意思就是通过 @Bean 注解，装配一个 EurekaAutoServiceRegistration 对象作为Spring的bean，而我们从名字就可以看出来EurekaClient的注册就是 EurekaAutoServiceRegistration 对象所进行的操作

同时，在这个方法上，也有这么一行 @ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled", matchIfMissing = true)

综上所述：我们可以看出来，EurekaClient的注册和两个配置项有关的，一个是 eureka.client.enabled ，另一个是 spring.cloud.service-registry.auto-registration.enabled ，只不过这两个配置默认都是true。这两个配置无论哪个我们手动配置成false，我们的服务都无法进行注册，测试自行做

另外还有一个原因：上图中不是提到了 EurekaAutoServiceRegistration类+@Bean注解 吗，那去看一下

可以看到 EurekaAutoServiceRegistration 类实现了Spring的 SmartLifecycle 接口，这个接口的作用是帮助一个类在作为Spring的Bean的时候，由Spring帮助我们自动进行一些和生命周期有关的工作，比如在初始化或者停止的时候进行一些操作。而我们最关心的 注册(register) 这个动作，就是在SmartLifecycle接口的 start() 方法实现里完成的

而上一步讲到，EurekaAutoServiceRegistration 类在 EurekaClientAutoConfiguration 类里恰好被配置成Spring的Bean，所以这里的 start() 方法是会自动被Spring调用的，我们不需要进行任何操作

总结

当我们引用了EurekaClient的依赖后，并且 eureka.client.enabled 和 spring.cloud.service-registry.auto-registration.enabled 两个开关不手动置为false，Spring就会自动帮助我们执行 EurekaAutoServiceRegistration 类里的 start() 方法，而注册的动作就是在该方法里完成的

所以，我们的EurekaClient工程，并不需要显式地在SpringBoot的启动类上标注 @EnableEurekaClient 注解

服务消费者

创建Maven模块，依赖如下：

  org.springframework.boot  spring-boot-starter-web  org.springframework.cloud  spring-cloud-starter-netflix-eureka-client

YAML配置如下：

server:  port: 8080spring:  application:    name: ORDER-SERVICEeureka:  client:    service-url:      # 向哪个eureka server进行服务拉取      defaultZone: http://localhost:10086/eureka

启动类如下：

import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.cloud.client.loadbalancer.LoadBalanced;import org.springframework.context.annotation.Bean;import org.springframework.web.client.RestTemplate;@SpringBootApplicationpublic class OrderApplication {    public static void main(String[] args) {        SpringApplication.run(OrderApplication.class, args);    }    /**     * RestTemplate 用来进行远程调用服务提供方的服务     * LoadBalanced 注解 是SpringCloud中的     *              此处作用：赋予RestTemplate负载均衡的能力 也就是在依赖注入时，只注入实例化时被@LoadBalanced修饰的实例     *              底层是 Spring的Qualifier注解，即为spring的原生操作     */    @Bean    @LoadBalanced    public RestTemplate restTemplate() {        return new RestTemplate();    }

@Qualifier 注解很重要：

@Autowired 默认是根据类型进行注入的，因此如果有多个类型一样的Bean候选者，则需要限定其中一个候选者，否则将抛出异常

@Qualifier 限定描述符除了能根据名字进行注入，更能进行更细粒度的控制如何选择候选者

@LoadBalanced很明显，”继承”了注解@Qualifier，RestTemplates通过@Autowired注入，同时被@LoadBalanced修饰，所以只会注入@LoadBalanced修饰的RestTemplate，也就是我们的目标RestTemplate

通过 RestTemplate +eureka 远程调用服务提供方中的服务

import com.zixieqing.order.mapper.OrderMapper;import com.zixieqing.order.pojo.Order;import com.zixieqing.order.pojo.User;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;import org.springframework.web.client.RestTemplate;@Servicepublic class OrderService {    @Autowired    private OrderMapper orderMapper;    @Autowired    private RestTemplate restTemplate;    public Order queryOrderById(Long orderId) {        // 1.查询订单        Order order = orderMapper.findById(orderId);        // 2、远程调用服务的url 此处直接使用服务名，不用ip+port        // 原因是底层有一个LoadBalancerInterceptor，里面有一个intercept()，后续玩负载均衡Ribbon会看到        String url = "http://USER-SERVICE/user/" + order.getUserId();        // 2.1、利用restTemplate调用远程服务，封装成user对象        User user = restTemplate.getForObject(url, User.class);        // 3、给oder设置user对象值        order.setUser(user);        // 4.返回        return order;    }}

不会玩 RestTemplate 用法的 戳这里

测试

依次启动eureka-server、user-service、order-service，然后将user-service做一下模拟集群即可，将user-service弄为模拟集群操作方式如下：不同版本IDEA操作有点区别，出入不大

再将复刻的use-service2也启动即可，启动之后点一下eureka-server的端口就可以在浏览器看到服务qingk

可以自行在服务提供方和服务消费方编写逻辑，去链接数据库，然后在服务消费方调用服务提供方的业务，最后访问自己controller中定义的路径和参数即可

Ribbon负载均衡Ribbon是什么？

Ribbon是Netflix发布的开源项目，Spring Cloud Ribbon是基于Netflix Ribbon实现的一套客户端负载均衡的框架

Ribbon属于哪种负载均衡？

LB负载均衡(Load Balance)是什么？

• 简单地说就是将用户的请求平摊的分配到多个服务上，从而达到系统的HA（高可用）
• 常见的负载均衡有软件Nginx，硬件 F5等

什么情况下需要负载均衡？

• 现在Java非常流行微服务，也就是所谓的面向服务开发，将一个项目拆分成了多个项目，其优点有很多，其中一个优点就是：将服务拆分成一个一个微服务后，我们很容易地来针对性的进行集群部署。例如订单模块用的人比较多，那就可以将这个模块多部署几台机器，来分担单个服务器的压力

• 这时候有个问题来了，前端页面请求的时候到底请求集群当中的哪一台？既然是降低单个服务器的压力，所以肯定全部机器都要利用起来，而不是说一台用着，其他空余着。这时候就需要用负载均衡了，像这种前端页面调用后端请求的，要做负载均衡的话，常用的就是Nginx

Ribbon和Nginx负载均衡的区别

• 当后端服务是集群的情况下，前端页面调用后端请求，要做负载均衡的话，常用的就是Nginx
• Ribbon主要是在“服务端内”做负载均衡，举例：订单后端服务 要调用 支付后端服务，这属于后端之间的服务调用，压根根本不经过页面，而后端支付服务是集群，这时候订单服务就需要做负载均衡来调用支付服务，记住是订单服务做负载均衡 “来调用” 支付服务

负载均衡分类

• 集中式LB：即在服务的消费方和提供方之间使用独立的LB设施(可以是硬件，如F5, 也可以是软件，如nginx)，由该设施负责把访问请求通过某种策略转发至服务的提供方
• 进程内LB：将LB逻辑集成到“消费方”，消费方从服务注册中心获知有哪些地址可用，然后自己再从这些地址中选择出一个合适的服务器

Ribbon负载均衡

• Ribbon就属于进程内LB，它只是一个类库，集成于服务消费方进程

Ribbon的流程

通过上图一定要明白一点：Ribbon一定是用在消费方，而不是服务的提供方！

Ribbon在工作时分成两步（这里以Eureka为例，consul和zk同样道理）：

• 第一步先选择 EurekaServer ，它优先选择在同一个区域内负载较少的server
• 第二步再根据用户指定的策略(轮询、随机、响应时间加权…..)，从server取到的服务注册列表中选择一个地址

请求怎么从服务名地址变为真实地址的？

只要引入了注册中心(Eureka、consul、zookeeper)，那Ribbon的依赖就在注册中心里面了，证明如下：

回到正题：为什么下面这样使用服务名就可以调到服务提供方的服务，即：请求 http://userservice/user/101 怎么变成的 http://localhost:8081 ？？因为它长得好看？

import com.zixieqing.order.mapper.OrderMapper;import com.zixieqing.order.pojo.Order;import com.zixieqing.order.pojo.User;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;import org.springframework.web.client.RestTemplate;@Servicepublic class OrderService {    @Autowired    private OrderMapper orderMapper;    @Autowired    private RestTemplate restTemplate;    public Order queryOrderById(Long orderId) {        // 1.查询订单        Order order = orderMapper.findById(orderId);        // 2、远程调用服务的url 此处直接使用服务名，不用ip+port        // 原因是底层有一个LoadBalancerInterceptor，里面有一个intercept()，后续玩负载均衡Ribbon会看到        String url = "http://USER-SERVICE/user/" + order.getUserId();        // 2.1、利用restTemplate调用远程服务，封装成user对象        User user = restTemplate.getForObject(url, User.class);        // 3、给oder设置user对象值        order.setUser(user);        // 4.返回        return order;    }}// RestTemplate做了下面操作，使用了 @Bean+@LoadBalanced    /**     * RestTemplate 用来进行远程调用服务提供方     * LoadBalanced 注解 是SpringCloud中的     *              此处作用：赋予RestTemplate负载均衡的能力 也就是在依赖注入时，只注入实例化时被@LoadBalanced修饰的实例     *              底层是 Spring的Qualifier注解，即为spring的原生操作     */    @Bean    @LoadBalanced    public RestTemplate restTemplate() {        return new RestTemplate();    }

想知道答案就得Debug了，而要Debug，就得找到 LoadBalancerInterceptor 类

LoadBalancerInterceptor类

然后对服务消费者进行Debug

问题的答案已经出来了：为什么使用服务名就可以调到服务提供方的服务，即：请求 http://userservice/user/101 怎么变成的 http://localhost:8081 ？？

• 原因就是使用了RibbonLoadBalancerClient+loadBalancer(默认是 ZoneAwareLoadBalance 从服务列表中选取服务)+IRule(默认是 RoundRobinRule 轮询策略选择某个服务)

总结

SpringCloudRibbon的底层采用了一个拦截器LoadBalancerInterceptor，拦截了RestTemplate发出的请求，对地址做了修改

负载均衡策略有哪些？

根据前面的铺垫，也知道了负载均衡策略就在 IRule 中，那就去看一下

转换一下：

ClientConfigEnabledRoundRobinRule：该策略较为特殊，我们一般不直接使用它。因为它本身并没有实现什么特殊的处理逻辑。一般都是可以通过继承他重写一些自己的策略，默认的choose()就实现了线性轮询机制

• BestAvailableRule：继承自ClientConfigEnabledRoundRobinRule，会先过滤掉由于多次访问故障而处于断路器跳闸状态的服务,然后选择一个并发量最小的服务，该策略的特性是可选出最空闲的实例

PredicateBasedRule：继承自ClientConfigEnabledRoundRobinRule，抽象策略，需要重写方法，然后自定义过滤规则

• AvailabilityFilteringRule：继承PredicateBasedRule，先过滤掉故障实例,再选择并发较小的实例。过滤掉的故障服务器是以下两种：
  1. 在默认情况下，这台服务器如果3次连接失败，这台服务器就会被设置为“短路”状态。短路状态将持续30秒，如果再次连接失败，短路的持续时间就会几何级地增加
  2. 并发数过高的服务器。如果一个服务器的并发连接数过高，配置了AvailabilityFilteringRule规则的客户端也会将其忽略。并发连接数的上限，可以由客户端的..ActiveConnectionsLimit 属性进行配置
• ZoneAvoidanceRule：继承PredicateBasedRule，默认规则，复合判断server所在区域的性能和server的可用性选择服务器

com.netflix.loadbalancer.RoundRobinRule：轮询 Ribbon的默认规则

• WeightedResponseTimeRule：对RoundRobinRule的扩展。为每一个服务器赋予一个权重值，服务器响应时间越长，其权重值越小，这个权重值会影响服务器的选择，即：响应速度越快的实例选择权重越大,越容易被选择
• ResponseTimeWeightedRule：对RoundRobinRule的扩展。响应时间加权

com.netflix.loadbalancer.RandomRule：随机

com.netflix.loadbalancer.StickyRule：这个基本也没人用

com.netflix.loadbalancer.RetryRule：先按照RoundRobinRule的策略获取服务,如果获取服务失败则在指定时间内会进行重试，从而获取可用的服务

ZoneAvoidanceRule：先复合判断server所在区域的性能和server的可用性选择服务器，再使用Zone对服务器进行分类，最后对Zone内的服务器进行轮询

自定义负载均衡策略

在前面已经知道了策略是 IRule ，所以就是改变了这个玩意而已

1、代码方式 ：服务消费者的启动类或重开config模块编写如下内容即可

@Beanpublic IRule randomRule(){    // new前面提到的那些rule对象即可，当然这里面也可以自行篡改策略逻辑返回    return new RandomRule();}

注： 此种方式是全局策略，即所有服务均采用这里定义的负载均衡策略

2、@RibbonClient注解：用法如下

/** * 在服务消费者的启动类中加入如下注解即可 如下注解指的是：调用 USER-SERVICE 服务时 使用MySelfRule负载均衡规则 * * 这里的MySelfRule可以弄为自定义逻辑的策略，也可以是前面提到的那些rule策略 */@RibbonClient(name = "USER-SERVICE",configuration=MySelfRule.class)

这种方式可以达到只针对某服务做负载均衡策略，但是：官方给出了明确警告 configuration=MySelfRule.class 自定义配置类一定不能放到@ComponentScan 所扫描的当前包下以及子包下，否则我们自定义的这个配置类就会被所有的Ribbon客户端所共享，达不到特殊化定制的目的了（也就是一旦被扫描到，RestTemplate直接不管调用哪个服务都会用指定的算法）

springboot项目当中的启动类使用了@SpringBootApplication注解，这个注解内部就有@ComponentScan注解，默认是扫描启动类包下所有的包，所以我们要达到定制化一定不要放在它能扫描到的地方

cloud中文官网：https://www.springcloud.cc/spring-cloud-greenwich.html#netflix-ribbon-starter

3、使用YAML配置文件方式 在服务消费方的yml配置文件中加入如下格式的内容即可

# 给某个微服务配置负载均衡规则，这里是user-service服务user-service:   ribbon:    # 负载均衡规则    NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RandomRule

注意，一般用默认的负载均衡规则，不做修改

Ribbon饿汉加载

Ribbon默认是采用懒加载，即第一次访问时才会去创建LoadBalanceClient，请求时间会很长。

而饿汉加载则会在项目启动时创建，降低第一次访问的耗时，通过下面配置开启饥饿加载：

ribbon:  eager-load:    # 开启负载均衡饿汉加载模式    enabled: true    # clients是一个String类型的List数组，多个时采用下面的 - xxxx服务 的形式，单个时直接使用 clients: 服务名 即可    clients:      - USER-SERVICE

Nacos注册中心

国内公司一般都推崇阿里巴巴的技术，比如注册中心，SpringCloudAlibaba也推出了一个名为Nacos的注册中心

Nacos 是阿里巴巴的产品，现在是 SpringCloud 中的一个组件。相比 Eureka 功能更加丰富，在国内受欢迎程度较高

安装Nacoswindows安装

GitHub中下载：https://github.com/alibaba/nacos/releases

下载好之后直接解压即可，但：别解压到有“中文路径”的地方

Nacos的默认端口是8848，若该端口被占用则关闭该进程 或 修改nacos中的默认端口(conf/application.properties)

启动Nacos：密码和账号均是 nacos

startup.cmd -m standalone-m modul 模式standalone单机

Linux安装

Nacos是基于Java开发的，所以需要JDK支持，因此Linux中需要有JDK环境

上传Linux版的JDK

# 解压tar -xvf jdk-8u144-linux-x64.tar.gz# 配置环境变量export JAVA_HOME=/usr/local/java# =JDK解压后的路径export PATH=$PATH:$JAVA_HOME/bin# 刷新环境变量source /etc/profile

上传Linux版的Nacos

# 解压tar -xvf nacos-server-1.4.1.tar.gz# 进入 nacos/bin 目录中，输入命令启动Nacossh startup.sh -m standalone# 有8848端口冲突和windows中一样方式解决

注册服务到Nacos中

拉取Nacos的依赖管理，服务端加入如下依赖

  com.alibaba.cloud  spring-cloud-alibaba-dependencies  2.2.5.RELEASE  pom  import

客户端依赖如下：

  com.alibaba.cloud  spring-cloud-starter-alibaba-nacos-discovery

注：不要有其他注册中心的依赖，如前面玩的Eureka，有的话注释掉

修改客户端的yml配置文件：

server:  port: 8081spring:  application:    name: USER-SERVICE  cloud:    nacos:      # Nacos服务器地址      server-addr: localhost:8848#eureka:#  client:#    # 去哪里拉取服务列表#    service-url:#      defaultZone: http://localhost:10086/eureka

启动之后，在 ip+port/nacos 就在Nacos控制台看到信息了

Nacos集群配置与负载均衡策略调整

1、集群配置：Nacos的服务多级存储模型和其他的不一样

就多了一个集群，不像其他的是 服务–> 实例

好处：微服务互相访问时，应该尽可能访问同集群实例，因为本地访问速度更快。当本集群内不可用时，才访问其它集群

配置服务集群：想要对哪个服务配置集群则在其yml配置文件中加入即可

server:  port: 8081  application:    name: USER-SERVICE  cloud:    nacos:      # Nacos服务器地址      server-addr: localhost:8848      # 配置集群名称，如：HZ，杭州      cluster-name: HZ

测试则直接将“服务提供者”复刻多份，共用同一集群名启动，然后再复刻修改集群名启动即可，如下面的：

2、负载均衡策略调整：前面玩Ribbon时已经知道了默认是轮询策略，而想要达到Nacos的 尽可能访问同集群实例，因为本地访问速度更快。当本集群内不可用时，才访问其它集群 的功能，则就需要调整负载均衡策略，配置如下：

USER-SERVICE:  ribbon:    # 单独对某个服务设置负载均衡策略#    NFLoadBalancerRuleClassName: com.netflix.loadbalancer.RoundRobinRule    # 改为Naocs的负载均衡策略    NFLoadBalancerRuleClassName: com.alibaba.cloud.nacos.ribbon.NacosRule

注： 再次说明前面提到的 ——> 负载均衡策略调整放在“服务消费方”

经过上面的配置之后，服务消费方去调用服务提供方的服务时，会优先选择和服务消费方同集群下的服务提供方的服务，若无法访问才跨集群访问其他集群下的服务提供方得到服务

• 小细节： 服务消费方访问同集群下服务提供方的服务时(提供方是集群，多实例)，选择这些实例中的哪一个服务时并不是采用轮询了，而是随机

另外的负载均衡策略就是Ribbon中的：

3、加权策略 ：服务器权重值越高，越容易被选择，所以能者多劳，性能好的服务器被访问的次数应该越多

权重值一般在 [0,10000] 之间。直接去Nacos的控制台中选择想要修改权重值的服务，点击“详情”即可修改

注： 当权重值为0时，代表此服务实例不会再被访问，类似于停机迭代

Nacos环境隔离

前面一节见到了Nacos的集群结构，但那只是较内的一层，Nacos不止是注册中心，也可以是数据中心

• namespace ：就是环境隔离，如 dev开发环境、test测试环境、prod生产环境。若没配置，则默认是public，在没有指定命名空间时都会默认从public这个命名空间拉取配置以及注册到该命名空间下的注册表中。什么是注册表在后续看源码时会说明
• group ：就是在namespace的基础上，再进行分组，就是平时理解的分组，如 将服务相关性强的分在一个组
• service —-> clusters —–> instances ：就是前面说的集群，服务 —-> 集群 ——> 实例

配置namespace： 注意事项如下

1. 同名的命名空间只能创建一个
2. 微服务间如果没有注册到一个命名空间下，无法使用OpenFeign指定服务名负载通信（服务拉取的配置文件不同命名空间不影响）。Feign是后面要玩的

在yml配置文件中进行环境隔离配置

spring:  cloud:    nacos:      server-addr: localhost:8848      cluster-name: HZ      # 环境隔离：即当前这个服务要注册到哪个命名空间环境去      # 值为在Nacos控制台创建命名空间时的id值，如下面的dev环境      namespace: e7144264-0bf4-4caa-a17d-0af8e81eac3a

Nacos临时与非临时实例

1、Nacos和Eureka的不同：不同在下图字体加粗的部分，加粗是Nacos具备而Eureka不具备的

临时实例： 由服务提供者主动给Nacos发送心跳情况，在规定时间内要是没有发送，则Nacos认为此服务挂了，就会从服务列表中踢掉（非亲儿子）

非临时实例/永久实例：由Nacos主动来询问服务是否还健康、活着(此种实例会让服务器压力变大)，若非临时实例挂了，Naocs并不会将其踢掉（亲儿子）

1. 临时实例：Nacos官网https://nacos.io/zh-cn/docs/open-api.html中的“服务发现”的“发送实例心跳”中可以看到源码是在什么地方找

   • 适合：流量激增时使用(高并发故增加更多实例)，后续流量下降了这些实例就可以不要了

   • 采用客户端心跳检测模式，心跳周期5秒

   • 心跳间隔超过15秒则标记为不健康

   • 心跳间隔超过30秒则从服务列表删除

2. 永久实例：

   • 适合：常备实例

   • 采用服务端主动健康检测方式

   • 周期为2000 + 5000，即[2000, 7000]毫秒内的随机数

   • 检测异常只会标记为不健康，不会删除

push：若是Nacos检测到有服务提供者挂了，就会主动给消费者发送服务变更的消息，然后服务消费者更新自己的服务缓存列表。这一步就会让服务列表更新很及时

• 此方式是Nacos具备而Eureka不具备的，Eureka只有pull操作，因此Eureka的缺点就是服务更新可能会不及时(在30s内，服务提供者变动了，个别挂了，而消费者中的服务缓存列表还是旧的，只能等到30s到了才去重新pull)

Nacos的服务发现分为两种模式：

• 模式一：主动拉取模式(push模式)，消费者定期主动从Nacos拉取服务列表并缓存起来，再服务调用时优先读取本地缓存中的服务列表
• 模式二：订阅模式(pull模式)，消费者订阅Nacos中的服务列表，并基于UDP协议来接收服务变更通知。当Nacos中的服务列表更新时，会发送UDP广播给所有订阅者

查看服务发现源码的地方：后续也会介绍

Nacos集群默认采用AP方式，当集群中存在非临时实例时，采用CP模式；Eureka采用AP方式

补充：CAP定理 这是分布式事务中的一个方法论

1. C即：Consistency 数据一致性。指的是：用户访问分布式系统中的任意节点，得到的数据必须一致
2. A即：Availability 可用性。指的是：用户访问集群中的任意健康节点，必须能得到响应，而不是超时或拒绝
3. P即：Partition Tolerance 分区容错性。指的是：由于某种原因导致系统中任意信息的丢失或失败都不能不会影响系统的继续独立运作

注： 分区容错性是必须满足的，数据一致性( C ）和 可用性（ A ）只满足其一即可，一般的搭配是如下的（即：取舍策略）：

1. CP保证数据的准确性
2. AP保证数据的及时性

既然CAP定理都整了，那就再加一个Base理论吧，这个理论是对CAP中C和A这两个矛盾点的调和和选择

1. BA 即：Basically Available 基本可用性。指的是：在发生故障的时候，可以允许损失“非核心部分”的可用性，保证系统正常运行，即保证核心部分可用
2. S 即：Soft State 软状态。指的是：允许系统的数据存在中间状态，只要不影响整个系统的运行就行
3. E 即：Eventual Consistency 最终一致性。指的是：无论以何种方式写入数据库 / 显示出来，都要保证系统最终的数据是一致的

2、配置临时实例与非临时实例：在需要的一方的yml配置文件中配置如下开关即可

spring:  cloud:    nacos:      server-addr: localhost:8848      cluster-name: HZ      # 默认为true，即临时实例      ephemeral: false

改完之后可以在Nacos控制台看到服务是否为临时实例

Nacos统一配置管理

统一配置管理： 将容易发生改变的配置单独弄出来，然后在后续需要变更时，直接去统一配置管理处进行更改，这样凡是依赖于这些配置的服务就可以统一被更新，而不用挨个服务更改配置，同时更改配置之后不用重启服务，直接实现热更新

Nacos和SpringCloud原生的config不一样，Nacos是将 注册中心+config 结合在一起了，而SpringCloud原生的是Eureka+config

1、设置Nacos配置管理

以上便是在Nacos中设置了统一配置。但是：项目/服务想要得到这些配置，那就得获取到这些配置，怎么办？

在前面说过SpringCloud中有两种yml的配置方式，一种是 application.yml ，一种是 bootstrap.yml ，这里就需要借助后者了，它是引导文件，优先级比前者高，会优先被加载，这样就可以先使用它加载到Nacos中的配置文件，然后再读取 application.yml ，从而完成Spring的那一套注册实例的事情

2、在需要读取Nacos统一配置的服务中引入如下依赖：

  com.alibaba.cloud  spring-cloud-starter-alibaba-nacos-config

3、resources下新建 bootstrap.yml，里面的配置内容如下

spring:  application:    # 服务名，对应在nacos中进行配置管理的data id的服务名    name: userservice  profiles:    # 环境，对应在nacos中进行配置管理的data id的环境    active: dev  cloud:    nacos:      # nacos服务器地址，需要知道去哪里拉取配置信息      server-addr: localhost:8848    config:      # 文件后缀，对应在nacos中进行配置管理的data id的后缀名      file-extension: yaml

经过上面的操作之后，以前需要单独在 application.yml 改的事情就不需要了，bootstrap.yml 配置的东西会去拉取nacos中的配置

4、设置热更新： 假如业务代码中有需要用到nacos中的配置信息，那nacos中的配置改变之后，不需要重启服务，自动更新。一共有两种方式

1. @RefreshScope+@Value注解： 在 @Value 注入的变量所在类上添加注解 @RefreshScope

2. @ConfigurationProperties 注解

然后在需要的地方直接注入对象即可

Nacos多环境共享配置

有时会遇到这样的情况：生产环境、开发环境、测试环境有些配置是相同的，这种应该不需要在每个环境中都配置，因此需要让这些相同的配置单独弄出来，然后实行共享

在前面一节中已经说到了一种Nacos的配置文件格式 即 服务名-环境.后缀，除了这种还有一种格式 即 服务名.后缀

因此：想要让环境配置共享，那么直接在Nacos控制台的配置中再加一个以 服务名.后缀名 格式命名的配置即可，如下：

其他的都不用动，要只是针对于项目中的yml，如 appilication.yml，那前面已经说了，会先读取Nacos中配置，然后和 application.yml 进行合并

但是：若项目本地的yml中、服务名.后缀、服务名-环境.后缀 中有相同的属性/配置时，优先级不一样，如下：

Nacos集群部署

windows和Linux都是一样的思路，集群部署的逻辑如下：

1、解压压缩包

2、进入nacos的conf目录，修改配置文件cluster.conf.example，重命名为cluster.conf，并添加要部署的集群ip+port，如下：

ip1:port1ip2:port2ip3:port3

3、然后修改conf/application.properties文件，添加数据库配置

# 告诉nacos数据库集群是MySQL，根据需要自定义spring.datasource.platform=mysql# 数据库的数量db.num=1# 数据库urldb.url.0=jdbc:mysql://127.0.0.1:3306/nacos?characterEncoding=utf8&connectTimeout=1000&socketTimeout=3000&autoReconnect=true&useUnicode=true&useSSL=false&serverTimezone=UTC# 数据库用户名db.user.0=root# 数据库密码db.password.0=88888

4、复制解压包，部署到不同服务器，然后改变每个解压包的端口，路径：conf/application.properties文件，例如：

# 第一个nacos节点server.port=8845# 第二个nacos节点server.port=8846# 第三个nacos节点server.port=8847

5、挨个启动nacos即可，进入到解压的nacos的bin目录中，执行如下命令即可

startup.cmd此命令告知：nacos默认就是集群启动，前面玩时加了 -m standalone 就是单机启动

5、使用Nginx做反向代理 ：修改conf/nginx.conf文件，配置如下：

upstream nacos-cluster {  server ip1:port1;  server ip2:port2;  server ip3:port3;}server {  listen       80;  server_name  localhost;  location /nacos {    proxy_pass http://nacos-cluster;  }}

6、代码中application.yml文件配置如下：

spring:  cloud:    nacos:      # Nacos地址，上一步Nginx中的 server_name+listen监听的端口      server-addr: localhost:80

7、访问 http://localhost/nacos 即可

• 注：浏览器默认就是80端口，而上面Nginx中监听的就是80，所以根据情况自行修改这里的访问路径

Nacos服务注册表结构是怎样的？

分析源码就在nacos官网下载的source.code：nacos-naming/controller/InstanceController#register(HttpServletRequest request)

Java代码中是使用了Map<String, Map>：每一个服务去注册到Nacos时，就会把信息组织并存入这个Map中

• key：是namespace_id，用于环境隔离
• value：是Map
  • key：是group，但是是使用group+serviceName组成的key
  • value：表示Service服务，这个Service又套了一个Map
    • key：就是集群名
    • value：就是Cluster对象，这里面又套了一个Set，这里面就是实例了

Nacos如何为何能抗住数十万服务注册压力？

源码在：nacos-naming/controller/InstanceController#register(HttpServletRequest request)中的serviceManager.registerInstance(namespaceId, serviceName, instance)里面

先看抗住压力的原因的结论：

• 在Nacos集群下，对于临时实例，服务注册时是将其丢给了一个ArrayBlockingQueue阻塞队列，然后就返回客户端，最后通过一个死循环利用线程池去执行阻塞队列中的任务(注册服务)，这就做到了异步操作
• 将服务更新情况同步给集群中的其他节点也是同样的原理，底层还是用了阻塞队列+线程池

具体的逻辑在 DistroConsistencyServiceImpl.put()中

public class DistroConsistencyServiceImpl {    @Override    public void put(String key, Record value) throws NacosException {// 异步服务注册 key是服务唯一id，value就是instances        onPut(key, value);        // 服务更强情况异步更新给集群下的另外节点        distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,                globalConfig.getTaskDispatchPeriod() / 2);    }    /**     * Put a new record.     *     * @param key   key of record     * @param value record     */    public void onPut(String key, Record value) {        // 判断是否是临时实例        if (KeyBuilder.matchEphemeralInstanceListKey(key)) {            // 封装 Instances 信息到 数据集：Datum            Datum datum = new Datum();            // value就是instances            datum.value = (Instances) value;            // key是服务的唯一id            datum.key = key;            // 加入当前修改时间            datum.timestamp.incrementAndGet();            // 数据存储 放入dataStore中            dataStore.put(key, datum);        }        if (!listeners.containsKey(key)) {            return;        }        // notifier这玩意儿 implements Runnable        notifier.addTask(key, DataOperation.CHANGE);    }    public class Notifier implements Runnable {        private ConcurrentHashMap services = new ConcurrentHashMap(10 * 1024);        /**         * 维护了一个阻塞队列         */        private BlockingQueue<Pair> tasks = new ArrayBlockingQueue(1024 * 1024);        /**         * Add new notify task to queue.         *         * @param datumKey data key         * @param action   action for data         */        public void addTask(String datumKey, DataOperation action) {            if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {                return;            }            if (action == DataOperation.CHANGE) {                services.put(datumKey, StringUtils.EMPTY);            }            // 将服务唯一id + 事件类型(CHANGE)放入了阻塞队列            tasks.offer(Pair.with(datumKey, action));        }                                @Override        public void run() {            Loggers.DISTRO.info("distro notifier started");            for (; ; ) { // 死循环                try {                    // 去阻塞队列中获取任务                    Pair pair = tasks.take();                    // 有任务就处理任务，更新服务列表；无任务就进入wait，所以此死循环不会导致CPU负载过高                    handle(pair);                } catch (Throwable e) {                    Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);                }            }        }                        /** * DistroConsistencyServiceImpl.Notifier类的 handle 方法：即 handle(pair) 中的逻辑 */private void handle(Pair pair) {            try {                String datumKey = pair.getValue0();                DataOperation action = pair.getValue1();                services.remove(datumKey);                int count = 0;                if (!listeners.containsKey(datumKey)) {                    return;                }                // 遍历，找到变化的service，这里的 RecordListener 就是 Service                for (RecordListener listener : listeners.get(datumKey)) {                    count++;                    try {                        // 如果是 CHANGE 事件                        if (action == DataOperation.CHANGE) {                            // 就更新服务列表                            listener.onChange(datumKey, dataStore.get(datumKey).value);                            continue;                        }                        // 如果是 DELETE 事件                        if (action == DataOperation.DELETE) {                            // 就根据服务ID删除从服务列表中删除服务                            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);            }        }    }}

因此能抗住压力的原因：

• 在Nacos集群下，对于临时实例，服务注册时是将其丢给了一个ArrayBlockingQueue阻塞队列，然后就返回客户端，最后通过一个死循环利用线程池去执行阻塞队列中的任务(注册服务)，这就做到了异步操作
• 将服务更新情况同步给集群中的其他节点也是同样的原理，底层还是用了阻塞队列+线程池

Nacos实例的并发读写问题

源码还是在：nacos-naming/controller/InstanceController#register(HttpServletRequest request)中的serviceManager.registerInstance(namespaceId, serviceName, instance)里面

具体思路：采用了同步锁+CopyOnWrite思想

• 并发读的解决方式 – CopyOnWrite思想：将原来的实例列表Map拷贝给了一个新的Map，然后对新的实例列表Map进行增删，最后将新的实例列表Map的引用给旧的实例列表Map
• 并发写的解决方式：
  • 在注册实例时，会使用synchronized同步锁对service进行加锁，不同service不影响，相同service通过锁排斥
  • 另外还有一个原因是：更新实例列表时，底层使用了线程池异步更新实例列表，但是线程池的线程数量为“1”

@Componentpublic class ServiceManager {    public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)            throws NacosException {        // 监听服务列表用到的key，服务唯一标识        // 如：com.alibaba.nacos.naming.iplist.ephemeral.public##DEFAULT_GROUP@@order-service        String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);        // 获取服务        Service service = getService(namespaceId, serviceName);        // 同步锁：解决并发写的问题        synchronized (service) {            // 1、获取要更新的实例列表            // addIPAddress中，会拷贝旧的实例列表，添加新实例到列表中 即：COPY            List instanceList = addIpAddresses(service, ephemeral, ips);            // 2、将更新后的数据封装到Instances对象            Instances instances = new Instances();            instances.setInstanceList(instanceList);            // 3、完成 注册表更新 以及 Nacos集群的数据同步(保证集群一致性)            // 在这里面 完成对实例状态更新后，会用新列表直接覆盖旧实例列表。而在更新过程中，旧实例列表不受影响，用户依然可以读取            consistencyService.put(key, instances);        }    }    private List addIpAddresses(Service service, boolean ephemeral, Instance... ips) throws NacosException {        return updateIpAddresses(service, UtilsAndCommons.UPDATE_INSTANCE_ACTION_ADD, ephemeral, ips);    }    /**     * 拷贝旧的实例列表，添加新实例到列表中     */    public List updateIpAddresses(Service service, String action, boolean ephemeral, Instance... ips)            throws NacosException {        // 根据namespaceId、serviceName获取当前服务的实例列表，返回值是Datum        // 第一次来，肯定是null        Datum datum = consistencyService                .get(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), ephemeral));        // 得到服务中旧的实例列表        List currentIPs = service.allIPs(ephemeral);        // 保存实例列表，key为ip地址，value是Instance对象        Map currentInstances = new HashMap(currentIPs.size());        // 创建Set集合，保存实例的instanceId        Set currentInstanceIds = Sets.newHashSet();        // 遍历旧实例列表        for (Instance instance : currentIPs) {            // 保存实例列表            currentInstances.put(instance.toIpAddr(), instance);            // 添加instanceId到set中            currentInstanceIds.add(instance.getInstanceId());        }        // 用来保存更新后的实例列表        Map instanceMap;        // 如果服务中已经有旧的数据        if (datum != null && null != datum.value) {            // 将旧实例列表与新实例列表进行比对、合并            instanceMap = setValid(((Instances) datum.value).getInstanceList(), currentInstances);        } else {            // 若服务中没有数据，则直接创建新的map            instanceMap = new HashMap(ips.length);        }        // 遍历新实例列表ips        for (Instance instance : ips) {            // 判断服务中是否包含要注册的实例的cluster信息            if (!service.getClusterMap().containsKey(instance.getClusterName())) {                // 如果不包含，创建新的cluster                Cluster cluster = new Cluster(instance.getClusterName(), service);                cluster.init();                // 将集群放入service的注册表                service.getClusterMap().put(instance.getClusterName(), cluster);                // ......记录日志            }            // 删除实例 or 新增实例            // 若是Remove删除事件类型            if (UtilsAndCommons.UPDATE_INSTANCE_ACTION_REMOVE.equals(action)) {                // 则通过实例ID删除实例                instanceMap.remove(instance.getDatumKey());            } else {                // 通过实例ID从旧实例列表中获取实例                Instance oldInstance = instanceMap.get(instance.getDatumKey());                if (oldInstance != null) {                    // 若旧实例列表中有这个实例 则将旧实例ID赋值给新实例ID                    instance.setInstanceId(oldInstance.getInstanceId());                } else {                    // 若旧实例列表中没有这个实例 则给新实例生成一个实例ID                    instance.setInstanceId(instance.generateInstanceId(currentInstanceIds));                }                // 实例ID为key、实例为value存入新实例列表                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()));        }        // 将instanceMap中的所有实例转为List返回        return new ArrayList(instanceMap.values());    }}

服务注册源码

Nacos的注册表结构是什么样的？

• Nacos是多级存储模型，最外层通过namespace来实现环境隔离，然后是group分组，分组下就是service服务，一个服务又可以分为不同的cluster集群，集群中包含多个instance实例。因此其注册表结构为一个Map，类型是：Map<String, Map>，

  外层key是namespace_id，内层key是group+serviceName.

  Service内部维护一个Map，结构是：Map，key是clusterName，值是集群信息

  Cluster内部维护一个Set集合，元素是Instance类型，代表集群中的多个实例。

Nacos如何保证并发写的安全性？

• 在注册实例时，会对service加锁，不同service之间本身就不存在并发写问题，互不影响。相同service时通过锁来互斥。并且，在更新实例列表时，是基于异步的线程池来完成，而线程池的线程数量为1.

问题延伸：Nacos是如何应对数十万服务的并发写请求？

• Nacos内部会将服务注册的任务放入阻塞队列，采用线程池异步来完成实例更新，从而提高并发写能力

Nacos如何避免并发读写的冲突？

• Nacos在更新实例列表时，会采用CopyOnWrite技术，首先将Old实例列表拷贝一份，然后更新拷贝的实例列表，再用更新后的实例列表来覆盖旧的实例列表。

客户端

流程如下：

NacosServiceRegistryAutoConfiguration

Nacos的客户端是基于SpringBoot的自动装配实现的，我们可以在nacos-discovery依赖：

spring-cloud-starter-alibaba-nacos-discovery-2.2.6.RELEASE.jar

这个包中找到Nacos自动装配信息：

可以看到，在NacosServiceRegistryAutoConfiguration这个类中，包含一个跟自动注册有关的Bean：

NacosAutoServiceRegistration

可以看到在初始化时，其父类AbstractAutoServiceRegistration也被初始化了

AbstractAutoServiceRegistration如图：

可以看到它实现了ApplicationListener接口，监听Spring容器启动过程中的事件

在监听到WebServerInitializedEvent（web服务初始化完成）的事件后，执行了bind 方法。

其中的bind方法如下：

public void bind(WebServerInitializedEvent event) {    // 获取 ApplicationContext    ApplicationContext context = event.getApplicationContext();    // 判断服务的 namespace,一般都是null    if (context instanceof ConfigurableWebServerApplicationContext) {        if ("management".equals(((ConfigurableWebServerApplicationContext) context)                                .getServerNamespace())) {            return;        }    }    // 记录当前 web 服务的端口    this.port.compareAndSet(0, event.getWebServer().getPort());    // 启动当前服务注册流程    this.start();}

其中的start方法流程：

public void start() {if (!isEnabled()) {if (logger.isDebugEnabled()) {logger.debug("Discovery Lifecycle disabled. Not starting");}return;}// 当前服务处于未运行状态时，才进行初始化if (!this.running.get()) {            // 发布服务开始注册的事件this.context.publishEvent(new InstancePreRegisteredEvent(this, getRegistration()));            // ☆☆☆☆开始注册☆☆☆☆register();if (shouldRegisterManagement()) {registerManagement();}            // 发布注册完成事件this.context.publishEvent(new InstanceRegisteredEvent(this, getConfiguration()));            // 服务状态设置为运行状态，基于AtomicBooleanthis.running.compareAndSet(false, true);}}

其中最关键的register()方法就是完成服务注册的关键，代码如下：

protected void register() {    this.serviceRegistry.register(getRegistration());}

此处的this.serviceRegistry就是NacosServiceRegistry：

NacosServiceRegistry

NacosServiceRegistry是Spring的ServiceRegistry接口的实现类，而ServiceRegistry接口是服务注册、发现的规约接口，定义了register、deregister等方法的声明。

而NacosServiceRegistry对register的实现如下：

@Overridepublic void register(Registration registration) {// 判断serviceId是否为空，也就是spring.application.name不能为空    if (StringUtils.isEmpty(registration.getServiceId())) {        log.warn("No service to register for nacos client...");        return;    }    // 获取Nacos的命名服务，其实就是注册中心服务    NamingService namingService = namingService();    // 获取 serviceId 和 Group    String serviceId = registration.getServiceId();    String group = nacosDiscoveryProperties.getGroup();// 封装服务实例的基本信息，如 cluster-name、是否为临时实例、权重、IP、端口等    Instance instance = getNacosInstanceFromRegistration(registration);    try {        // 开始注册服务        namingService.registerInstance(serviceId, group, instance);        log.info("nacos registry, {} {} {}:{} register finished", group, serviceId,                 instance.getIp(), instance.getPort());    }    catch (Exception e) {        if (nacosDiscoveryProperties.isFailFast()) {            log.error("nacos registry, {} register failed...{},", serviceId,                      registration.toString(), e);            rethrowRuntimeException(e);        }        else {            log.warn("Failfast is false. {} register failed...{},", serviceId,                     registration.toString(), e);        }    }}

可以看到方法中最终是调用NamingService的registerInstance方法实现注册的

而NamingService接口的默认实现就是NacosNamingService

NacosNamingService

NacosNamingService提供了服务注册、订阅等功能

其中registerInstance就是注册服务实例，源码如下：

@Overridepublic void registerInstance(String serviceName, String groupName, Instance instance) throws NacosException {    // 检查超时参数是否异常。心跳超时时间(默认15秒)必须大于心跳周期(默认5秒)    NamingUtils.checkInstanceIsLegal(instance);    // 拼接得到新的服务名，格式为：groupName@@serviceId    String groupedServiceName = NamingUtils.getGroupedName(serviceName, groupName);    // 判断是否为临时实例，默认为 true。    if (instance.isEphemeral()) { // 这里面的两行代码很关键        // 如果是临时实例，需要定时向 Nacos 服务发送心跳 ---------- 涉及临时实例的心跳检测        BeatInfo beatInfo = beatReactor.buildBeatInfo(groupedServiceName, instance);        // 添加心跳任务        beatReactor.addBeatInfo(groupedServiceName, beatInfo);    }    // 发送注册服务实例的请求    serverProxy.registerService(groupedServiceName, groupName, instance);}

最终，由NacosProxy的registerService方法，完成服务注册

public void registerService(String serviceName, String groupName, Instance instance) throws NacosException {// 组织请求参数    final Map params = new HashMap(16);    params.put(CommonParams.NAMESPACE_ID, namespaceId);    params.put(CommonParams.SERVICE_NAME, serviceName);    params.put(CommonParams.GROUP_NAME, groupName);    params.put(CommonParams.CLUSTER_NAME, 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", JacksonUtils.toJson(instance.getMetadata()));// 通过POST请求将上述参数，发送到 /nacos/v1/ns/instance    reqApi(UtilAndComs.nacosUrlInstance, params, HttpMethod.POST);}

这里提交的信息就是Nacos服务注册接口需要的完整参数，核心参数有：

• namespace_id：环境
• service_name：服务名称
• group_name：组名称
• cluster_name：集群名称
• ip: 当前实例的ip地址
• port: 当前实例的端口

服务端

服务端流程图：

官网下载源码：进入 naming-nacos/com/alibaba/nacos/naming/controllers/InstanceController#register(HttpServletRequest request)

@CanDistro@PostMapping@Secured(parser = NamingResourceParser.class, action = ActionTypes.WRITE)public String register(HttpServletRequest request) throws Exception {    // 尝试获取namespaceId    final String namespaceId = WebUtils        .optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);    // 尝试获取serviceName，其格式为 group_name@@service_name    final String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);    NamingUtils.checkServiceNameFormat(serviceName);// 解析出实例信息，封装为Instance对象    final Instance instance = parseInstance(request);// 注册实例    serviceManager.registerInstance(namespaceId, serviceName, instance);    return "ok";}

进入serviceManager.registerInstance(namespaceId, serviceName, instance)

ServiceManager

这里面的东西在前面并发读写的解决方式中见过了

这里面的流程一句话来说就是：先获取旧的实例列表，然后把新的实例信息与旧的做对比、合并，新的实例就添加，老的实例同步ID。然后返回最新的实例列表

registerInstance方法就是注册服务实例的方法：

/** * 注册服务实例 * * Register an instance to a service in AP mode. * * 
This method creates service or cluster silently if they don't exist. * * @param namespaceId id of namespace * @param serviceName service name * @param instance    instance to register * @throws Exception any error occurred in the process */public void registerInstance(String namespaceId, String serviceName, Instance instance) throws NacosException {    // 创建一个空的service（如果是第一次来注册实例，要先创建一个空service出来，放入注册表）    // 此时不包含实例信息    createEmptyService(namespaceId, serviceName, instance.isEphemeral());    // 拿到创建好的service    Service service = getService(namespaceId, serviceName);    // 拿不到则抛异常    if (service == null) {        throw new NacosException(NacosException.INVALID_PARAM,                                 "service not found, namespace: " + namespaceId + ", service: " + serviceName);    }    // 添加要注册的实例到service中    addInstance(namespaceId, serviceName, instance.isEphemeral(), instance);}

创建好了服务，接下来就要添加实例到服务中：

/** * 添加实例到服务中 *  * Add instance to service. * * @param namespaceId namespace * @param serviceName service name * @param ephemeral   whether instance is ephemeral * @param ips         instances * @throws NacosException nacos exception */public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)    throws NacosException {    // 监听服务列表用到的key    // 服务唯一标识，例如：com.alibaba.nacos.naming.iplist.ephemeral.public##DEFAULT_GROUP@@order-service    String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);    // 获取服务    Service service = getService(namespaceId, serviceName);    // 同步锁，避免并发修改的安全问题    synchronized (service) {        // 1、获取要更新的实例列表        // addIPAddress中，会拷贝旧的实例列表，添加新实例到列表中 即：COPY        List instanceList = addIpAddresses(service, ephemeral, ips);        // 2、将更新后的数据封装到Instances对象        Instances instances = new Instances();        instances.setInstanceList(instanceList);        // 3、完成 注册表更新 以及 Nacos集群的数据同步(保证集群一致性)        // 在这里面 完成对实例状态更新后，会用新列表直接覆盖旧实例列表。而在更新过程中，旧实例列表不受影响，用户依然可以读取        consistencyService.put(key, instances);    }}

最后就要更新服务的实例 列表了

private List addIpAddresses(Service service, boolean ephemeral, Instance... ips) throws NacosException {    return updateIpAddresses(service, UtilsAndCommons.UPDATE_INSTANCE_ACTION_ADD, ephemeral, ips);}/** * 拷贝旧的实例列表，添加新实例到列表中 */public List updateIpAddresses(Service service, String action, boolean ephemeral, Instance... ips)    throws NacosException {    // 根据namespaceId、serviceName获取当前服务的实例列表，返回值是Datum    // 第一次来，肯定是null    Datum datum = consistencyService        .get(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), ephemeral));    // 得到服务中旧的实例列表    List currentIPs = service.allIPs(ephemeral);    // 保存实例列表，key为ip地址，value是Instance对象    Map currentInstances = new HashMap(currentIPs.size());    // 创建Set集合，保存实例的instanceId    Set currentInstanceIds = Sets.newHashSet();    // 遍历旧的实例列表    for (Instance instance : currentIPs) {        // 保存实例列表        currentInstances.put(instance.toIpAddr(), instance);        // 添加instanceId到set中        currentInstanceIds.add(instance.getInstanceId());    }    // 用来保存更新后的实例列表    Map instanceMap;    // 如果服务中已经有旧的数据    if (datum != null && null != datum.value) {        // 将旧的实例列表与新的实例列表进行比对        instanceMap = setValid(((Instances) datum.value).getInstanceList(), currentInstances);    } else {        // 若服务中没有数据，则直接创建新的map        instanceMap = new HashMap(ips.length);    }    // 遍历新的实例列表ips    for (Instance instance : ips) {        // 判断服务中是否包含要注册的实例的cluster信息        if (!service.getClusterMap().containsKey(instance.getClusterName())) {            // 如果不包含，创建新的cluster            Cluster cluster = new Cluster(instance.getClusterName(), service);            cluster.init();            // 将集群放入service的注册表            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());        }        // 删除实例 or 新增实例        // 若是Remove删除事件类型        if (UtilsAndCommons.UPDATE_INSTANCE_ACTION_REMOVE.equals(action)) {            // 则通过实例ID删除实例            instanceMap.remove(instance.getDatumKey());        } else {            // 通过实例ID从旧实例列表中获取实例            Instance oldInstance = instanceMap.get(instance.getDatumKey());            if (oldInstance != null) {                // 若旧实例列表中有这个实例 则将旧实例ID赋值给新实例ID                instance.setInstanceId(oldInstance.getInstanceId());            } else {                // 若旧实例列表中没有这个实例 则给新实例生成一个实例ID                instance.setInstanceId(instance.generateInstanceId(currentInstanceIds));            }            // 实例ID为key、实例为value存入新实例列表            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()));    }    // 将instanceMap中的所有实例转为List返回    return new ArrayList(instanceMap.values());}

Nacos集群一致性

在上一节中，在完成本地服务列表更新后，Nacos又实现了集群一致性更新，调用的是:

consistencyService.put(key, instances);

/** * 添加实例到服务中 *  * Add instance to service. * * @param namespaceId namespace * @param serviceName service name * @param ephemeral   whether instance is ephemeral * @param ips         instances * @throws NacosException nacos exception */public void addInstance(String namespaceId, String serviceName, boolean ephemeral, Instance... ips)    throws NacosException {    // 监听服务列表用到的key    // 服务唯一标识，例如：com.alibaba.nacos.naming.iplist.ephemeral.public##DEFAULT_GROUP@@order-service    String key = KeyBuilder.buildInstanceListKey(namespaceId, serviceName, ephemeral);    // 获取服务    Service service = getService(namespaceId, serviceName);    // 同步锁，避免并发修改的安全问题    synchronized (service) {        // 1、获取要更新的实例列表        // addIPAddress中，会拷贝旧的实例列表，添加新实例到列表中 即：COPY        List instanceList = addIpAddresses(service, ephemeral, ips);        // 2、将更新后的数据封装到Instances对象        Instances instances = new Instances();        instances.setInstanceList(instanceList);        // 3、完成 注册表更新 以及 Nacos集群的数据同步(保证集群一致性)        // 在这里面 完成对实例状态更新后，会用新列表直接覆盖旧实例列表。而在更新过程中，旧实例列表不受影响，用户依然可以读取        consistencyService.put(key, instances);    }}

这里的ConsistencyService接口，代表集群一致性的接口，有很多中不同实现：

进入DelegateConsistencyServiceImpl来看：

@Overridepublic void put(String key, Record value) throws NacosException {    // 根据实例是否是临时实例，判断委托对象    mapConsistencyService(key).put(key, value);}

其中的mapConsistencyService(key)方法就是选择委托方式：

private ConsistencyService mapConsistencyService(String key) {    // 判断是否是临时实例：    // 是，选择 ephemeralConsistencyService，也就是 DistroConsistencyServiceImpl    // 否，选择 persistentConsistencyService，也就是 PersistentConsistencyServiceDelegateImpl    return KeyBuilder.matchEphemeralKey(key) ? ephemeralConsistencyService : persistentConsistencyService;}

默认情况下，所有实例都是临时实例，因此关注DistroConsistencyServiceImpl即可

DistroConsistencyServiceImpl

这里面的逻辑在前面“Nacos如何抗住数十万服务注册压力”中见过了的，但是没弄全

@Overridepublic void put(String key, Record value) throws NacosException {    // 异步服务注册 key是服务的唯一id，value就是instances    onPut(key, value);    // 服务更强情况异步更新给集群下的另外节点    distroProtocol.sync(new DistroKey(key, KeyBuilder.INSTANCE_LIST_KEY_PREFIX), DataOperation.CHANGE,                        globalConfig.getTaskDispatchPeriod() / 2);}

onPut 更新本地实例列表

@DependsOn("ProtocolManager")@org.springframework.stereotype.Service("distroConsistencyService")public class DistroConsistencyServiceImpl implements EphemeralConsistencyService, DistroDataProcessor {public void onPut(String key, Record value) {        // 判断是否是临时实例        if (KeyBuilder.matchEphemeralInstanceListKey(key)) {            // 封装 Instances 信息到 数据集：Datum            Datum datum = new Datum();            // value就是instances            datum.value = (Instances) value;            // key是服务的唯一id            datum.key = key;            // 加入当前修改时间            datum.timestamp.incrementAndGet();            // 数据存储 放入dataStore中            dataStore.put(key, datum);        }        if (!listeners.containsKey(key)) {            return;        }        // notifier这玩意儿 implements Runnable        notifier.addTask(key, DataOperation.CHANGE);    }            public class Notifier implements Runnable {        private ConcurrentHashMap services = new ConcurrentHashMap(10 * 1024);        /**         * 维护了一个阻塞队列         */        private BlockingQueue<Pair> tasks = new ArrayBlockingQueue(1024 * 1024);        /**         * Add new notify task to queue.         *         * @param datumKey data key         * @param action   action for data         */        public void addTask(String datumKey, DataOperation action) {            if (services.containsKey(datumKey) && action == DataOperation.CHANGE) {                return;            }            if (action == DataOperation.CHANGE) {                services.put(datumKey, StringUtils.EMPTY);            }            // 将服务唯一id + 事件类型(CHANGE)放入了阻塞队列            tasks.offer(Pair.with(datumKey, action));        }    }}

Notifier异步更新

Notifier是一个Runnable，通过一个单线程的线程池来不断从阻塞队列中获取任务，执行服务列表的更新

@DependsOn("ProtocolManager")@org.springframework.stereotype.Service("distroConsistencyService")public class DistroConsistencyServiceImpl implements EphemeralConsistencyService, DistroDataProcessor {    public class Notifier implements Runnable {        private ConcurrentHashMap services = new ConcurrentHashMap(10 * 1024);        /**         * 维护了一个阻塞队列         */        private BlockingQueue<Pair> tasks = new ArrayBlockingQueue(1024 * 1024);                @Override        public void run() {            Loggers.DISTRO.info("distro notifier started");            for (; ; ) { // 死循环                try {                    // 去阻塞队列中获取任务                    Pair pair = tasks.take();                    // 有任务就处理任务，更新服务列表；无任务就进入wait，所以此死循环不会导致CPU负载过高                    handle(pair);                } catch (Throwable e) {                    Loggers.DISTRO.error("[NACOS-DISTRO] Error while handling notifying task", e);                }            }        }        private void handle(Pair pair) {            try {                String datumKey = pair.getValue0();                DataOperation action = pair.getValue1();                services.remove(datumKey);                int count = 0;                if (!listeners.containsKey(datumKey)) {                    return;                }                // 遍历，找到变化的service，这里的 RecordListener就是 Service                for (RecordListener listener : listeners.get(datumKey)) {                    count++;                    try {                        // 如果是 CHANGE 事件                        if (action == DataOperation.CHANGE) {                            // 就更新服务列表                            listener.onChange(datumKey, dataStore.get(datumKey).value);                            continue;                        }                        // 如果是 DELETE 事件                        if (action == DataOperation.DELETE) {                            // 就根据服务ID删除从服务列表中删除服务                            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);            }        }    }}

onChange 覆盖实例列表

上一节中 listener.onChange(datumKey, dataStore.get(datumKey).value); 进去，选择Service的onChange()

@JsonInclude(Include.NON_NULL)public class Service extends com.alibaba.nacos.api.naming.pojo.Service implements Record, RecordListener {    @Override    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.0D) {                instance.setWeight(0.01D);            }        }                // 更新实例列表        updateIPs(value.getInstanceList(), KeyBuilder.matchEphemeralInstanceListKey(key));                recalculateChecksum();    }}

updateIPs 的逻辑如下：

@JsonInclude(Include.NON_NULL)public class Service extends com.alibaba.nacos.api.naming.pojo.Service implements Record, RecordListener {    /** * 更新实例列表 *     * Update instances.     *     * @param instances instances     * @param ephemeral whether is ephemeral instance     */    public void updateIPs(Collection instances, boolean ephemeral) {        // key是cluster，值是集群下的Instance集合        Map<String, List> ipMap = new HashMap(clusterMap.size());        // 获取服务的所有cluster名称        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;                }                // 判断实例是否包含clusterName，没有的话用默认cluster                if (StringUtils.isEmpty(instance.getClusterName())) {                    // DEFAULT_CLUSTER_NAME = "DEFAULT"                    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();                    getClusterMap().put(instance.getClusterName(), cluster);                }                // 获取当前cluster实例的集合，不存在则创建新的                List clusterIPs = ipMap.get(instance.getClusterName());                if (clusterIPs == null) {                    clusterIPs = new LinkedList();                    ipMap.put(instance.getClusterName(), clusterIPs);                }                // 添加新的实例到 Instance 集合                clusterIPs.add(instance);            } catch (Exception e) {                Loggers.SRV_LOG.error("[NACOS-DOM] failed to process ip: " + instance, e);            }        }        for (Map.Entry<String, List> entry : ipMap.entrySet()) {            //make every ip mine            List entryIPs = entry.getValue();            // 将实例集合更新到 clusterMap（注册表）            clusterMap.get(entry.getKey()).updateIps(entryIPs, ephemeral);        }        setLastModifiedMillis(System.currentTimeMillis());        // 发布服务变更的通知消息        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());    }}

上面的 clusterMap.get(entry.getKey()).updateIps(entryIPs, ephemeral); 就是在更新实例列表，进入 updateIps(entryIPs, ephemeral) 即可看到逻辑

public class Cluster extends com.alibaba.nacos.api.naming.pojo.Cluster implements Cloneable {    /** * 更新实例列表 *     * Update instance list.     *     * @param ips       instance list     * @param ephemeral whether these instances are ephemeral     */    public void updateIps(List ips, boolean ephemeral) {        // 获取旧实例列表        Set toUpdateInstances = ephemeral ? ephemeralInstances : persistentInstances;        HashMap oldIpMap = new HashMap(toUpdateInstances.size());        for (Instance ip : toUpdateInstances) {            oldIpMap.put(ip.getDatumKey(), ip);        }        // 更新实例列表        List updatedIPs = updatedIps(ips, oldIpMap.values());        if (updatedIPs.size() > 0) {            for (Instance ip : updatedIPs) {                Instance oldIP = oldIpMap.get(ip.getDatumKey());                // do not update the ip validation status of updated ips                // because the checker has the most precise result                // Only when ip is not marked, don't we update the health status of IP:                if (!ip.isMarked()) {                    ip.setHealthy(oldIP.isHealthy());                }                if (ip.isHealthy() != oldIP.isHealthy()) {                    // ip validation status updated                    Loggers.EVT_LOG.info("{} {SYNC} IP-{} {}:{}@{}", getService().getName(),                            (ip.isHealthy() ? "ENABLED" : "DISABLED"), ip.getIp(), ip.getPort(), getName());                }                if (ip.getWeight() != oldIP.getWeight()) {                    // ip validation status updated                    Loggers.EVT_LOG.info("{} {SYNC} {IP-UPDATED} {}->{}", getService().getName(), oldIP.toString(),                            ip.toString());                }            }        }        // 检查新加入实例的状态        List newIPs = subtract(ips, oldIpMap.values());        if (newIPs.size() > 0) {            Loggers.EVT_LOG                    .info("{} {SYNC} {IP-NEW} cluster: {}, new ips size: {}, content: {}", getService().getName(),                            getName(), newIPs.size(), newIPs.toString());            for (Instance ip : newIPs) {                HealthCheckStatus.reset(ip);            }        }        // 移除要删除的实例        List deadIPs = subtract(oldIpMap.values(), ips);        if (deadIPs.size() > 0) {            Loggers.EVT_LOG                    .info("{} {SYNC} {IP-DEAD} cluster: {}, dead ips size: {}, content: {}", getService().getName(),                            getName(), deadIPs.size(), deadIPs.toString());            for (Instance ip : deadIPs) {                HealthCheckStatus.remv(ip);            }        }        toUpdateInstances = new HashSet(ips);        // 直接覆盖旧实例列表        if (ephemeral) {            ephemeralInstances = toUpdateInstances;        } else {            persistentInstances = toUpdateInstances;        }    }}

Nacos集群一致性

@Componentpublic class DistroProtocol {    /** * 同步数据到其他远程服务器 *     * Start to sync data to all remote server.     *     * @param distroKey distro key of sync data     * @param action    the action of data operation     */    public void sync(DistroKey distroKey, DataOperation action, long delay) {        // 遍历 Nacos 集群中除自己以外的其它节点        for (Member each : memberManager.allMembersWithoutSelf()) {            DistroKey distroKeyWithTarget = new DistroKey(distroKey.getResourceKey(), distroKey.getResourceType(),                    each.getAddress());            // Distro同步任务            DistroDelayTask distroDelayTask = new DistroDelayTask(distroKeyWithTarget, action, delay);            // 交给线程池去执行            distroTaskEngineHolder.getDelayTaskExecuteEngine().addTask(distroKeyWithTarget, distroDelayTask);            if (Loggers.DISTRO.isDebugEnabled()) {                Loggers.DISTRO.debug("[DISTRO-SCHEDULE] {} to {}", distroKey, each.getAddress());            }        }    }}

distroTaskEngineHolder.getDelayTaskExecuteEngine() 的返回值是 NacosDelayTaskExecuteEngine，它维护了一个线程池，并且接收任务，执行任务。执行任务的方法为processTasks()方法

public class NacosDelayTaskExecuteEngine extends AbstractNacosTaskExecuteEngine {    protected void processTasks() {        Collection