注册中心

CAP原则

CAP原则又称CAP定理，指的是在一个分布式系统中，一致性（Consistency）、可用性（Availability）、分区容错性（Partition tolerance），这三个要素最多只能同时实现两点，不可能三者兼顾。

BASE

BASE是Basically Available（基本可用）、Soft state（软状态）和 Eventually consistent（最终一致性）三个短语的简写。

BASE是对 CAP 中一致性和可用性权衡的结果，其来源于对大规模互联网系统分布式实践的结论，是基于CAP定理逐步演化而来的，其核心思想是即使无法做到强一致性（Strong consistency），但每个应用都可以根据自身的业务特点，采用适当的方式来使系统达到最终一致性（Eventual consistency）。接下来我们着重对BASE中的三要素进行详细讲解。基本可用：指分布式系统在出现不可预知故障的时候，允许损失部分可用性。

常见注册中心比较

一个基本的注册中心需要以下 4 个基本的功能：

• 注册服务实例信息
• 心跳机制
• 剔除失败的服务实例信息
• 查询服务实例信息操作

zookeeper

zk 本身并不是为了做注册中心的，不过其提供的通用树状存储结构和 znode 机可以间接完成服务发现的必要功能。比如我们有 2 个服务 a 和 b

/
├ a
┆ ├ a1
┆ └ a2
└ b
    └ b1

这样存储，可以通过查询 a 节点，获取服务 a 下面的实例信息。

在 zk 中，可以在使用临时节点创建 a1、a2、b1 这样的用来存储服务实例信息的节点，当服务实例关闭或者通信异常时，zookeeper 可以自动删除这些临时节点，这样就实现了剔除机制。

zk，一旦服务挂掉，zk感知到以及通知其他服务的时效性，服务注册到zk之后通知到其他服务的时效性，leader挂掉之后可用性是否会出现短暂的问题，为了去换取一致性

注册机制：客户端主动创建临时节点

心跳机制：因为创建的是临时节点，依靠 zk 本身的会话机制

剔除机制：会话失效后，临时节点自动剔除

查询机制：使用 zk 协议去查询节点

Eureka

image-20210912133439362

相比于 zookeeper 来说，Eureka 是专门用来做注册中心的，本身提供了注册中心需要的所有的功能。其提供了 SDK 和 HTTP 接口来访问 Eureka Server.

其部分 API 如下，更多的查看 Eureka REST operations

如果不想使用 HTTP 接口，也可以直接使用 Eureka 提供的 Java SDK

Eureka 更侧重于 AP，其通过自我保护机制，可以在网络异常的情况下，保留大部分节点信息，来防止雪崩的情况

如果 Eureka 服务器检测到比预期数量多的注册客户端以不合适的方式终止了它们的连接，并且同时等待驱逐，它们将进入自我保护模式。 这样做是为了确保灾难性网络事件不会清除 eureka 注册表数据，并将其向下传播到所有客户端。

自我保护机制的工作机制是：如果在15分钟内超过85%的客户端节点都没有正常的心跳，那么Eureka就认为客户端与注册中心出现了网络故障，Eureka Server自动进入自我保护机制，此时会出现以下几种情况：

1. Eureka Server不再从注册列表中移除因为长时间没收到心跳而应该过期的服务。
2. Eureka Server仍然能够接受新服务的注册和查询请求，但是不会被同步到其它节点上，保证当前节点依然可用。
3. 当网络稳定时，当前Eureka Server新的注册信息会被同步到其它节点中。

具体可见 Server Self Preservation Mode

当 Eureka 进入自我保护机制的情况下，会造成服务实例无法剔除的情况，Client 在查询的时候可能查询到已经挂掉的实例信息。

Eureka 是 peer-to-peer 模式，可能还没同步数据过去，结果自己就死了，此时还是可以继续从别的机器上拉取注册表，但是看到的就不是最新的数据了，但是保证了可用性，强一致，最终一致性

注册机制：客户端主动创建节点信息（使用 SDK 或者 HTTP 接口）

心跳机制：客户端主动维持上报（使用 SDK 或者 HTTP 接口，默认 30s 上报一次）

剔除机制：未收到客户端 3 次心跳后，服务端主动删除

查询机制：客户端主动查询节点信息（使用 SDK 或者 HTTP 接口）

Consul

关于 Consul 和其他注册中心的对比，因为 Consul 本身出了文档这里不在赘叙 Consul VS Other

image-20210912135836507

Consul 本身提供了 Go SDK 和 HTTP 接口, 其中包括服务注册、健康检查、服务查询、kv 操作等功能的 API, 虽然没有提供其他的语言的官方 SDK, 但也有一些个人去封装了。或许可以使用非官方的或者自己封装 HTTP 接口。

相对于 Eureka，Consul 提供了多种心跳机制，包括：

1. Script + Interval
2. HTTP + Interval
3. TCP + Interval
4. Time to Live (TTL)
5. Docker + Interval
6. gRPC + Interval
7. H2ping + Interval
8. Alias

注册机制：客户端主动创建节点信息（使用 SDK 或者 HTTP 接口）

心跳机制：服务端根据你采用心跳机制对客户端进行心跳测试（和 Eureka、zk 不同，这里是服务端向客户端发起）

剔除机制：服务端未成功检测到客户端心跳反应后，服务端主动删除

查询机制：客户端主动查询节点信息（使用 SDK 或者 HTTP 接口）

Nacos

Nacos 支持基于 DNS 和基于 RPC 的服务发现。服务提供者使用 原生SDK、OpenAPI、或一个独立的Agent TODO注册 Service 后，服务消费者可以使用DNS TODO 或HTTP&API查找和发现服务。

Nacos 提供对服务的实时的健康检查，阻止向不健康的主机或服务实例发送请求。Nacos 支持传输层 (PING 或 TCP)和应用层 (如 HTTP、MySQL、用户自定义）的健康检查。 对于复杂的云环境和网络拓扑环境中（如 VPC、边缘网络等）服务的健康检查，Nacos 提供了 agent 上报模式和服务端主动检测2种健康检查模式。Nacos 还提供了统一的健康检查仪表盘，帮助您根据健康状态管理服务的可用性及流量。

image

image-20210912143100127

image-20210912143531008

注册机制：客户端主动创建节点信息（使用 SDK 或者 HTTP 接口）

心跳机制：客户端主动维持上报（使用 SDK 或者 HTTP 接口，默认 30s 上报一次）

剔除机制：未收到客户端 3 次心跳后，服务端主动删除

查询机制：客户端主动查询节点信息（使用 SDK 或者 HTTP 接口）

Spring Cloud 是如何实现服务治理的

Spring Cloud Commons 之服务治理浅析

Spring 在设计的时候，通常会考虑方便扩展和消除样板代码，在 Spring Clond 同样存在这样的设计。

在 Spring Cloud 体系中，Spring Cloud Commons 是最重要的一个项目，其中定义了服务注册、服务发现、负载均衡相关的接口以及一些公共组件，通过看这个项目，我们可以简单的理解一下 Spring Cloud 注册发现的核心流程。

Spring Clond Commons 项目中提供了如下的项目结构（在这里省略了部分代码文件和结构）

└── src
    ├── main
    │   ├── java
    │   │   └── org
    │   │       └── springframework
    │   │           └── cloud
    │   │               ├── client
    │   │               │   ├── DefaultServiceInstance.java
    │   │               │   ├── ServiceInstance.java  Spring Cloud 对服务实例信息的定义
    │   │               │   ├── discovery 服务发现相关
    │   │               │   │   ├── DiscoveryClient.java
    │   │               │   │   ├── EnableDiscoveryClient.java
    │   │               │   │   ├── EnableDiscoveryClientImportSelector.java
    │   │               │   │   ├── ManagementServerPortUtils.java
    │   │               │   │   ├── ReactiveDiscoveryClient.java
    │   │               │   │   ├── composite
    │   │               │   │   │   ├── CompositeDiscoveryClient.java
    │   │               │   │   │   ├── CompositeDiscoveryClientAutoConfiguration.java
    │   │               │   │   │   └── reactive
    │   │               │   │   │       ├── ReactiveCompositeDiscoveryClient.java
    │   │               │   │   │       └── ReactiveCompositeDiscoveryClientAutoConfiguration.java
    │   │               │   │   ├── health 健康检查相关
    │   │               │   │       ├── DiscoveryClientHealthIndicator.java
    │   │               │   │       ├── DiscoveryClientHealthIndicatorProperties.java
    │   │               │   │       ├── DiscoveryCompositeHealthContributor.java
    │   │               │   │       ├── DiscoveryHealthIndicator.java
    │   │               │   │       └── reactive
    │   │               │   │           ├── ReactiveDiscoveryClientHealthIndicator.java
    │   │               │   │           ├── ReactiveDiscoveryCompositeHealthContributor.java
    │   │               │   │           └── ReactiveDiscoveryHealthIndicator.java
    │   │               │   ├── loadbalancer 这下面是负载均衡相关逻辑
    │   │               │   └── serviceregistry 服务注册相关
    │   │               │       ├── AbstractAutoServiceRegistration.java
    │   │               │       ├── AutoServiceRegistration.java
    │   │               │       ├── AutoServiceRegistrationAutoConfiguration.java
    │   │               │       ├── AutoServiceRegistrationConfiguration.java
    │   │               │       ├── AutoServiceRegistrationProperties.java
    │   │               │       ├── Registration.java
    │   │               │       ├── ServiceRegistry.java
    │   │               │       ├── ServiceRegistryAutoConfiguration.java
    │   │               ├── commons
    │   │                   ├── httpclient http 工厂类，在配置中可以选择使用 Apache Http 还是 OKHttp
    │   │                   │   ├── ApacheHttpClientFactory.java
    │   │                   │   └── OkHttpClientFactory.java
    │   │                   └── util
    │   │                      ├── IdUtils.java   通过这工具类来生成实例 id
    │   │                      └── InetUtils.java Spring Cloud 就是通过这个工具类是获取服务项目的 ip 地址的
    │   └── resources
    │       └── META-INF
    │           ├── additional-spring-configuration-metadata.json
    │           └── spring.factories
    └── test
        ├── java 测试相关代码

在项目结构中可以看出各个部分对应的源码，在服务治理中，首先是服务信息 ServiceInstance , 其中包括

• 服务名 ServiceId 这个就是我们类似的 xxx-server （spring.application.name）
• 服务实例唯一标识符 InstanceId
• host
• port
• 一些扩展信息 metadata, 这个主要用来提供给三方实现增加以下扩展信息

// 为了缩短篇幅，删除了一些注释
public interface ServiceInstance {

    default String getInstanceId() {
        return null;
    }

    String getServiceId();

    String getHost();

    int getPort();

    boolean isSecure();

    URI getUri();

    Map<String, String> getMetadata();

    default String getScheme() {
        return null;
    }

}

服务注册

Registration 是 Spring Cloud 提供的一个注册实现

public interface Registration extends ServiceInstance {
    // 这里面是真没有代码
}

服务注册的实际接口是 ServiceRegistry

public interface ServiceRegistry<R extends Registration> {

    /**
     * Registers the registration. A registration typically has information about an
     * instance, such as its hostname and port.
     * @param registration registration meta data
     */
    void register(R registration);

    /**
     * Deregisters the registration.
     * @param registration registration meta data
     */
    void deregister(R registration);

    /**
     * Closes the ServiceRegistry. This is a lifecycle method.
     */
    void close();

    /**
     * Sets the status of the registration. The status values are determined by the
     * individual implementations.
     * @param registration The registration to update.
     * @param status The status to set.
     * @see org.springframework.cloud.client.serviceregistry.endpoint.ServiceRegistryEndpoint
     */
    void setStatus(R registration, String status);

    /**
     * Gets the status of a particular registration.
     * @param registration The registration to query.
     * @param <T> The type of the status.
     * @return The status of the registration.
     * @see org.springframework.cloud.client.serviceregistry.endpoint.ServiceRegistryEndpoint
     */
    <T> T getStatus(R registration);

}

通过实现 ServiceRegistry 即可完成一个简单服务注册功能

服务发现

在 discovery 下存在两个服务发现定义接口 DiscoveryClient 和 ReactiveDiscoveryClient

其提供了如下功能：

1. 获取所有的服务名称
2. 根据服务名称获取对应的服务实例列表

public interface DiscoveryClient extends Ordered {

    /**
     * Default order of the discovery client.
     */
    int DEFAULT_ORDER = 0;

    /**
     * A human-readable description of the implementation, used in HealthIndicator.
     * @return The description.
     */
    String description();

    /**
     * Gets all ServiceInstances associated with a particular serviceId.
     * @param serviceId The serviceId to query.
     * @return A List of ServiceInstance.
     */
    List<ServiceInstance> getInstances(String serviceId);

    /**
     * @return All known service IDs.
     */
    List<String> getServices();

    /**
     * Default implementation for getting order of discovery clients.
     * @return order
     */
    @Override
    default int getOrder() {
        return DEFAULT_ORDER;
    }

}

通过实现 DiscoveryClient 即可完成服务发现

健康检测

ReactiveDiscoveryClientHealthIndicator 提供了健康检测功能

1. 从 DiscoveryClient 中获取所有的服务名列表
2. 根据服务名列表获取对应的服务实例列表
3. 对每个实例进行健康检测，如果响应成功则 UP 否则为 DOWN

public class ReactiveDiscoveryClientHealthIndicator
        implements ReactiveDiscoveryHealthIndicator, Ordered, ApplicationListener<InstanceRegisteredEvent<?>> {

    private final ReactiveDiscoveryClient discoveryClient;

    private final DiscoveryClientHealthIndicatorProperties properties;

    private final Log log = LogFactory.getLog(ReactiveDiscoveryClientHealthIndicator.class);

    private AtomicBoolean discoveryInitialized = new AtomicBoolean(false);

    private int order = Ordered.HIGHEST_PRECEDENCE;

    public ReactiveDiscoveryClientHealthIndicator(ReactiveDiscoveryClient discoveryClient,
            DiscoveryClientHealthIndicatorProperties properties) {
        this.discoveryClient = discoveryClient;
        this.properties = properties;
    }

    @Override
    public void onApplicationEvent(InstanceRegisteredEvent<?> event) {
        if (this.discoveryInitialized.compareAndSet(false, true)) {
            this.log.debug("Discovery Client has been initialized");
        }
    }

    @Override
    public Mono<Health> health() {
        if (this.discoveryInitialized.get()) {
            return doHealthCheck();
        }
        else {
            return Mono.just(
                    Health.status(new Status(Status.UNKNOWN.getCode(), "Discovery Client not initialized")).build());
        }
    }

    private Mono<Health> doHealthCheck() {
        // @formatter:off
        return Mono.justOrEmpty(this.discoveryClient)
                .flatMapMany(ReactiveDiscoveryClient::getServices)
                .collectList()
                .defaultIfEmpty(emptyList())
                .map(services -> {
                    ReactiveDiscoveryClient client = this.discoveryClient;
                    String description = (this.properties.isIncludeDescription())
                            ? client.description() : "";
                    return Health.status(new Status("UP", description))
                            .withDetail("services", services).build();
                })
                .onErrorResume(exception -> {
                    this.log.error("Error", exception);
                    return Mono.just(Health.down().withException(exception).build());
                });
        // @formatter:on
    }

    @Override
    public String getName() {
        return discoveryClient.description();
    }

    @Override
    public int getOrder() {
        return this.order;
    }

    public void setOrder(int order) {
        this.order = order;
    }

}

通过上面的接口定义和自带的健康检测逻辑可以看出做一个服务治理需要实现的最简单的逻辑

1. 实现 ServiceRegistry 功能
2. 实现 DiscoveryClient 功能

Spring Cloud Consul 实现

实现 ServiceRegistry 功能

在 Spring Cloud Consul 中，首先自定义了 Registration 的实现

其中 NewService 为 Consul 定义的一些服务实例信息

public class ConsulRegistration implements Registration {

    private final NewService service;

    private ConsulDiscoveryProperties properties;

    public ConsulRegistration(NewService service, ConsulDiscoveryProperties properties) {
        this.service = service;
        this.properties = properties;
    }

    public NewService getService() {
        return this.service;
    }

    protected ConsulDiscoveryProperties getProperties() {
        return this.properties;
    }

    public String getInstanceId() {
        return getService().getId();
    }

    public String getServiceId() {
        return getService().getName();
    }

    @Override
    public String getHost() {
        return getService().getAddress();
    }

    @Override
    public int getPort() {
        return getService().getPort();
    }

    @Override
    public boolean isSecure() {
        return this.properties.getScheme().equalsIgnoreCase("https");
    }

    @Override
    public URI getUri() {
        return DefaultServiceInstance.getUri(this);
    }

    @Override
    public Map<String, String> getMetadata() {
        return getService().getMeta();
    }

}

NewService

其包含了服务的基本信息和 Consul 本身提供一些特有功能如：Tags、Check

// 删除了通用的 getter、setter、toString 方法
public class NewService {
  @SerializedName("ID")
  private String id;
  @SerializedName("Name")
  private String name;
  @SerializedName("Tags")
  private List<String> tags;
  @SerializedName("Address")
  private String address;
  @SerializedName("Meta")
  private Map<String, String> meta;
  @SerializedName("Port")
  private Integer port;
  @SerializedName("EnableTagOverride")
  private Boolean enableTagOverride;
  @SerializedName("Check")
  private NewService.Check check;
  @SerializedName("Checks")
  private List<NewService.Check> checks;

  public NewService() {
  }

  public static class Check {
    @SerializedName("Script")
    private String script;
    @SerializedName("DockerContainerID")
    private String dockerContainerID;
    @SerializedName("Shell")
    private String shell;
    @SerializedName("Interval")
    private String interval;
    @SerializedName("TTL")
    private String ttl;
    @SerializedName("HTTP")
    private String http;
    @SerializedName("Method")
    private String method;
    @SerializedName("Header")
    private Map<String, List<String>> header;
    @SerializedName("TCP")
    private String tcp;
    @SerializedName("Timeout")
    private String timeout;
    @SerializedName("DeregisterCriticalServiceAfter")
    private String deregisterCriticalServiceAfter;
    @SerializedName("TLSSkipVerify")
    private Boolean tlsSkipVerify;
    @SerializedName("Status")
    private String status;
    @SerializedName("GRPC")
    private String grpc;
    @SerializedName("GRPCUseTLS")
    private Boolean grpcUseTLS;

    public Check() {
    }
  }
}

ConsulServiceRegistry 实现 ServiceRegistry

public class ConsulServiceRegistry implements ServiceRegistry<ConsulRegistration> {

    private static Log log = LogFactory.getLog(ConsulServiceRegistry.class);

    private final ConsulClient client;

    private final ConsulDiscoveryProperties properties;

    private final TtlScheduler ttlScheduler;

    private final HeartbeatProperties heartbeatProperties;

    public ConsulServiceRegistry(ConsulClient client, ConsulDiscoveryProperties properties, TtlScheduler ttlScheduler,
            HeartbeatProperties heartbeatProperties) {
        this.client = client;
        this.properties = properties;
        this.ttlScheduler = ttlScheduler;
        this.heartbeatProperties = heartbeatProperties;
    }

    @Override
    public void register(ConsulRegistration reg) {
        log.info("Registering service with consul: " + reg.getService());
        try {
            // 同样是通过 consul 提供的 api 接口进行服务注册
            this.client.agentServiceRegister(reg.getService(), this.properties.getAclToken());
            NewService service = reg.getService();
            if (this.heartbeatProperties.isEnabled() && this.ttlScheduler != null && service.getCheck() != null
                    && service.getCheck().getTtl() != null) {
                this.ttlScheduler.add(reg.getInstanceId());
            }
        }
        catch (ConsulException e) {
            if (this.properties.isFailFast()) {
                log.error("Error registering service with consul: " + reg.getService(), e);
                ReflectionUtils.rethrowRuntimeException(e);
            }
            log.warn("Failfast is false. Error registering service with consul: " + reg.getService(), e);
        }
    }

    @Override
    public void deregister(ConsulRegistration reg) {
        if (this.ttlScheduler != null) {
            this.ttlScheduler.remove(reg.getInstanceId());
        }
        if (log.isInfoEnabled()) {
            log.info("Deregistering service with consul: " + reg.getInstanceId());
        }
        this.client.agentServiceDeregister(reg.getInstanceId(), this.properties.getAclToken());
    }

    @Override
    public void close() {

    }

    @Override
    public void setStatus(ConsulRegistration registration, String status) {
        if (status.equalsIgnoreCase(OUT_OF_SERVICE.getCode())) {
            this.client.agentServiceSetMaintenance(registration.getInstanceId(), true);
        }
        else if (status.equalsIgnoreCase(UP.getCode())) {
            this.client.agentServiceSetMaintenance(registration.getInstanceId(), false);
        }
        else {
            throw new IllegalArgumentException("Unknown status: " + status);
        }

    }

  // 服务实例状态
    @Override
    public Object getStatus(ConsulRegistration registration) {
        String serviceId = registration.getServiceId();
        Response<List<Check>> response = this.client.getHealthChecksForService(serviceId,
                HealthChecksForServiceRequest.newBuilder().setQueryParams(QueryParams.DEFAULT).build());
        List<Check> checks = response.getValue();

        for (Check check : checks) {
            if (check.getServiceId().equals(registration.getInstanceId())) {
                if (check.getName().equalsIgnoreCase("Service Maintenance Mode")) {
                    return OUT_OF_SERVICE.getCode();
                }
            }
        }

        return UP.getCode();
    }
}

ConsulDiscoveryClient 实现 DiscoveryClient

在发现逻辑中也是通过 consul 提供的 api 接口进行查询

public class ConsulDiscoveryClient implements DiscoveryClient {

    private final ConsulClient client;

    private final ConsulDiscoveryProperties properties;

    public ConsulDiscoveryClient(ConsulClient client, ConsulDiscoveryProperties properties) {
        this.client = client;
        this.properties = properties;
    }

    @Override
    public String description() {
        return "Spring Cloud Consul Discovery Client";
    }

    @Override
    public List<ServiceInstance> getInstances(final String serviceId) {
        return getInstances(serviceId, new QueryParams(this.properties.getConsistencyMode()));
    }

    public List<ServiceInstance> getInstances(final String serviceId, final QueryParams queryParams) {
        List<ServiceInstance> instances = new ArrayList<>();

        addInstancesToList(instances, serviceId, queryParams);

        return instances;
    }

    private void addInstancesToList(List<ServiceInstance> instances, String serviceId, QueryParams queryParams) {
        HealthServicesRequest.Builder requestBuilder = HealthServicesRequest.newBuilder()
                .setPassing(this.properties.isQueryPassing()).setQueryParams(queryParams)
                .setToken(this.properties.getAclToken());
        String queryTag = this.properties.getQueryTagForService(serviceId);
        if (queryTag != null) {
            requestBuilder.setTag(queryTag);
        }
        HealthServicesRequest request = requestBuilder.build();
        Response<List<HealthService>> services = this.client.getHealthServices(serviceId, request);

        for (HealthService service : services.getValue()) {
            instances.add(new ConsulServiceInstance(service, serviceId));
        }
    }

    public List<ServiceInstance> getAllInstances() {
        List<ServiceInstance> instances = new ArrayList<>();

        Response<Map<String, List<String>>> services = this.client
                .getCatalogServices(CatalogServicesRequest.newBuilder().setQueryParams(QueryParams.DEFAULT).build());
        for (String serviceId : services.getValue().keySet()) {
            addInstancesToList(instances, serviceId, QueryParams.DEFAULT);
        }
        return instances;
    }

    @Override
    public List<String> getServices() {
        CatalogServicesRequest request = CatalogServicesRequest.newBuilder().setQueryParams(QueryParams.DEFAULT)
                .setToken(this.properties.getAclToken()).build();
        return new ArrayList<>(this.client.getCatalogServices(request).getValue().keySet());
    }

    @Override
    public int getOrder() {
        return this.properties.getOrder();
    }

}

总结

简要的 Spring Cloud Consul 的服务治理逻辑大致如此，当然 Spring Cloud Consul 还要处理大量的细节，代码还是很多的

在 Spring Cloud 体系中 Consul 并不提供服务请求转发的功能，只是提供对服务信息的保存、查询、健康检测剔除功能

参考

1. Consul 官方介绍 https://www.consul.io/docs/intro
2. Spring Cloud Consul https://github.com/spring-cloud/spring-cloud-consul
3. Spring Cloud Commons https://github.com/spring-cloud/spring-cloud-commons
4. Nacos 文档 https://nacos.io
5. https://github.com/Netflix/eureka