亲宝软件园·资讯

展开

Spring Cloud进阶篇之Eureka原理分析

毛毛向前冲 人气:5

前言

之前写了几篇Spring Cloud的小白教程,相信看过的朋友对Spring Cloud中的一些应用有了简单的了解,写小白篇的目的就是为初学者建立一个基本概念,让初学者在学习的道路上建立一定的基础。

从今天开始,我会持续更新几篇Spring Cloud的进阶教程。

Eureka简介

Eureka是Netflix开发的服务发现框架,本身就是一个基于REST的服务。Spring Cloud将它集成在其子项目spring-cloud-netflix中,用来实现服务的注册与发现功能。

Eureka总体架构图

Eureka组件介绍

  • 服务注册中心集群

分别部署在IDC1、IDC2、IDC3中心

  • 服务提供者

服务提供者一个部署在IDC1,一个部署在IDC3

  • 服务消费者

服务消费者一个部署在IDC1,一个部署在IDC2

组件之间的调用关系

服务提供者

  • 启动服务:服务提供者会向服务注册中心发起Register请求,注册服务。
  • 运行过程中:服务提供者会定时向注册中心发送Renew心跳,告诉它“我还活着”。
  • 停止服务提供:服务提供者会向服务注册中心发送Cancel请求,告诉它清空当前服务注册信息。

服务消费者

  • 启动后:从服务注册中心拉取服务注册信息。
  • 运行过程中:定时更新服务注册信息。
  • 发起远程调用:
  • - 服务消费者会从服务注册中心选择同机房的服务提供者,然后发起远程调用,只有同机房的服务提供者宕机才会去选择其他机房的服务提供者。
  • 如果服务消费者发现同机房没有服务提供者,则会按照负载均衡算法 选择其他机房的服务提供者,然后发起远程调用。

注册中心

  • 启动后:从其他节点拉取服务注册信息
  • 运行过程中:
  • - 定时运行Evict任务,定时清理没有按时发送Renew的服务提供者,这里的清理会将非常正常停止、网络异常等其他因素引起的所有服务。
  • 接收到的Register、Renew、Cancel请求,都会同步到其他的注册中心节点。

Eureka Server会通过Register、Renew、Get Registry等接口提供服务的注册、发现和心跳检测等。

Eureka Client是一个java客户端,用于简化与Eureka Server的交互,客户端本身也内置了负载均衡器(默认使用round-robin方式),在启动后会向Eureka Server发送心跳检测,默认周期为30s,Eureka Server如果在多个心跳周期内没有接收到Eureka client的某一个节点的心跳请求,Eureka Server会从服务注册中心清理到对应的Eureka Client的服务节点(默认90s)。

数据结构

服务存储的数据结构可以简单的理解为是一个两层的HashMap结构(为了保证线程安全使用的ConcurrentHashMap),具体的我们可以查看源码中的AbstractInstanceRegistry类:

private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry = new ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>();

第一层ConcurrentHashMap的key=spring.application.name,也就是应用名称,value为ConcurrentHashMap。

第二层ConcurrentHashMap的key=instanceId,也就是服务的唯一实例id,value为Lease对象,也就是具体的服务。Lease其实就是对InstanceInfo的包装,里面保存了实例信息、服务注册的时间等。具体的我们可以查看InstanceInfo源码。

数据存储过程

Eureka是通过REST接口对外提供服务的。

这里我以注册为例(ApplicationResource),首先将PeerAwareInstanceRegistry的实例注入到ApplicationResource的成员变量的registry里。

  • ApplicationResource接收到请求后,对调用registry.register()方法。
@POST
    @Consumes({"application/json", "application/xml"})
    public Response addInstance(InstanceInfo info,
                                @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) {
        logger.debug("Registering instance {} (replication={})", info.getId(), isReplication);
        // validate that the instanceinfo contains all the necessary required fields
        if (isBlank(info.getId())) {
            return Response.status(400).entity("Missing instanceId").build();
        } else if (isBlank(info.getHostName())) {
            return Response.status(400).entity("Missing hostname").build();
        } else if (isBlank(info.getIPAddr())) {
            return Response.status(400).entity("Missing ip address").build();
        } else if (isBlank(info.getAppName())) {
            return Response.status(400).entity("Missing appName").build();
        } else if (!appName.equals(info.getAppName())) {
            return Response.status(400).entity("Mismatched appName, expecting " + appName + " but was " + info.getAppName()).build();
        } else if (info.getDataCenterInfo() == null) {
            return Response.status(400).entity("Missing dataCenterInfo").build();
        } else if (info.getDataCenterInfo().getName() == null) {
            return Response.status(400).entity("Missing dataCenterInfo Name").build();
        }

        // handle cases where clients may be registering with bad DataCenterInfo with missing data
        DataCenterInfo dataCenterInfo = info.getDataCenterInfo();
        if (dataCenterInfo instanceof UniqueIdentifier) {
            String dataCenterInfoId = ((UniqueIdentifier) dataCenterInfo).getId();
            if (isBlank(dataCenterInfoId)) {
                boolean experimental = "true".equalsIgnoreCase(serverConfig.getExperimental("registration.validation.dataCenterInfoId"));
                if (experimental) {
                    String entity = "DataCenterInfo of type " + dataCenterInfo.getClass() + " must contain a valid id";
                    return Response.status(400).entity(entity).build();
                } else if (dataCenterInfo instanceof AmazonInfo) {
                    AmazonInfo amazonInfo = (AmazonInfo) dataCenterInfo;
                    String effectiveId = amazonInfo.get(AmazonInfo.MetaDataKey.instanceId);
                    if (effectiveId == null) {
                        amazonInfo.getMetadata().put(AmazonInfo.MetaDataKey.instanceId.getName(), info.getId());
                    }
                } else {
                    logger.warn("Registering DataCenterInfo of type {} without an appropriate id", dataCenterInfo.getClass());
                }
            }
        }

        registry.register(info, "true".equals(isReplication));
        return Response.status(204).build();  // 204 to be backwards compatible
    }
  • AbstractInstanceRegistry在register方法里完成对服务信息的存储。
public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
        try {
            read.lock();
            Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
            REGISTER.increment(isReplication);
            if (gMap == null) {
                final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
                gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
                if (gMap == null) {
                    gMap = gNewMap;
                }
            }
            Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
            // Retain the last dirty timestamp without overwriting it, if there is already a lease
            if (existingLease != null && (existingLease.getHolder() != null)) {
                Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
                Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
                logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);

                // this is a > instead of a >= because if the timestamps are equal, we still take the remote transmitted
                // InstanceInfo instead of the server local copy.
                if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
                    logger.warn("There is an existing lease and the existing lease's dirty timestamp {} is greater" +
                            " than the one that is being registered {}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
                    logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");
                    registrant = existingLease.getHolder();
                }
            } else {
                // The lease does not exist and hence it is a new registration
                synchronized (lock) {
                    if (this.expectedNumberOfClientsSendingRenews > 0) {
                        // Since the client wants to register it, increase the number of clients sending renews
                        this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 1;
                        updateRenewsPerMinThreshold();
                    }
                }
                logger.debug("No previous lease information found; it is new registration");
            }
            Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
            if (existingLease != null) {
                lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
            }
            gMap.put(registrant.getId(), lease);
            synchronized (recentRegisteredQueue) {
                recentRegisteredQueue.add(new Pair<Long, String>(
                        System.currentTimeMillis(),
                        registrant.getAppName() + "(" + registrant.getId() + ")"));
            }
            // This is where the initial state transfer of overridden status happens
            if (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) {
                logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the "
                                + "overrides", registrant.getOverriddenStatus(), registrant.getId());
                if (!overriddenInstanceStatusMap.containsKey(registrant.getId())) {
                    logger.info("Not found overridden id {} and hence adding it", registrant.getId());
                    overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
                }
            }
            InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());
            if (overriddenStatusFromMap != null) {
                logger.info("Storing overridden status {} from map", overriddenStatusFromMap);
                registrant.setOverriddenStatus(overriddenStatusFromMap);
            }

            // Set the status based on the overridden status rules
            InstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant, existingLease, isReplication);
            registrant.setStatusWithoutDirty(overriddenInstanceStatus);

            // If the lease is registered with UP status, set lease service up timestamp
            if (InstanceStatus.UP.equals(registrant.getStatus())) {
                lease.serviceUp();
            }
            registrant.setActionType(ActionType.ADDED);
            recentlyChangedQueue.add(new RecentlyChangedItem(lease));
            registrant.setLastUpdatedTimestamp();
            invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
            logger.info("Registered instance {}/{} with status {} (replication={})",
                    registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);
        } finally {
            read.unlock();
        }
    }

从源码中不难看出存储的数据结构是双层的HashMap。

Eureka还实现了二级缓存来保证即将对外传输的服务信息,

  • 一级缓存:本质还是HashMap,没有过期时间,保存服务信息的对外输出的数据结构。

    private final ConcurrentMap<Key, Value> readOnlyCacheMap = new ConcurrentHashMap<Key, Value>();
  • 二级缓存:是guava的缓存,包含失效机制,保存服务信息的对外输出的数据结构。

    private final LoadingCache<Key, Value> readWriteCacheMap;
  • 缓存的更新:

  • 删除二级缓存:

  • - client端发送register、renew、cancel请求并更新register注册表之后会删除二级缓存;

  • server端自身的Evict任务剔除服务后会删除二级缓存;

  • 二级缓存本事设置的失效机制(指的是guava实现的readWriteCacheMap),

  • 加载二级缓存:

  • - client发送Get registry请求后,如果二级缓存中没有,就会触发guava的load机制,从registry中获取原始的服务信息后进行加工处理,然后放入二级缓存中;

  • server端更新一级缓存的时候,如果二级缓存没有数据也会触发guava的load机制;

  • 更新一级缓存:

  • - server端内置了一个time task会定时将二级缓存中的数据同步到一级缓存中,这其中包括了删除和更新。

缓存的机制可以查看ResponseCacheImpl源码。

Eureka的数据结构简单总结为:

服务注册机制

服务注册中心、服务提供者、服务消费者在启动后都会向服务注册中心发起注册服务的请求(前提是配置了注册服务)。

注册中心接到register请求后:

  • 将服务信息保存到registry中;

  • 更新队列,将该事件添加到更新队列中,给Eureka client增量同步服务信息使用;

  • 清空二级缓存,用于保证数据的一致性;(即清空的是:readWriteCacheMap

  • 更新阈值;

  • 同步服务信息;

服务续约

服务注册后,要定时发送续约请求(心跳检查),证明我还活着,不要清空我的服务信息,定时时间默认30s,可以通过配置:eureka.instance.lease-renewal-interval-in-seconds来修改。

注册中心接收到续约请求后(renew):

  • 更新服务对象的最近续约时间(lastUpdateTimestamp);
  • 将信息同步给其他的节点;

服务注销

正常的服务停止之前会发送注销服务请求,通知注册中心我要下线了。

注册中心接收到注销请求后(cancel):

  • 将服务信息从registry中删除;
  • 更新队列;
  • 清空二级缓存;
  • 更新阈值;
  • 同步信息给其他节点;

说明:只有服务正常停止才会发送cancel请求,非正常停止的会通过Eureka Server的主动剔除机制进行删除。

服务剔除

服务剔除其实是一个兜底的方案,目的就是解决非正常情况下的服务宕机或其他因素导致不能发送cancel请求的服务信息清理的策略。

服务剔除分为:

  • 判断剔除条件
  • 找出过期服务
  • 清理过期服务

剔除条件:

  • 关闭自我保护
  • 自我保护如果开启,会先判断是server还是client出现问题,如果是client的问题就会进行删除;

自我保护机制:Eureka的自我保护机制是为了防止误杀服务提供的一种保护机制。Eureka的自我保护机制认为如果有大量的服务都续约失败,则认为自己出现了问题(例如:自己断网了),也就不剔除了。反之,则是它人的问题,就进行剔除。

自我保护的阈值分为server和client,如果超出阈值就是表示大量服务可用,部分服务不可用,这判定为client端出现问题。如果未超出阈值就是表示大量服务不可用,则判定是自己出现了问题。

阈值的计算:

  • 自我保护阈值 = 服务总数 * 每分钟续约数 * 自我保护阈值因子;
  • 每分钟续约数 = (60s / 客户端续约时间);

过期服务:

找出过期服务会遍历所有的服务,判断上次续约时间距离当前时间大于阈值就标记为过期,同时会将这些过期的服务保存的过期的服务集合中。

剔除服务:

剔除服务之前会先计算要是剔除的服务数量,然后遍历过期服务,通过洗牌算法确保每次都公平的选择出要剔除的服务,然后进行剔除。

执行剔除服务后:

  • 从register中删除服务信息;
  • 更新队列;
  • 清空二级缓存,保证数据的一致性;

服务获取

Eureka Client服务的获取都是从缓存中获取,如果缓存中没有,就加载数据到缓存中,然后在从缓存中取。服务的获取方式分为全量同步和增量同步两种。

registry中只保存数据结构,缓存中存ready的服务信息

  • 先读取一级缓存
  • 先判断是否开启一级缓存
  • 如果开启一级缓存,就从一级缓存中取,如果一级缓存中没有,则从二级缓存中取;
  • 如果没有开启一级缓存,则直接从二级缓存中取;
  • 再读取二级缓存
  • 如果二级缓存中存在,则直接返回;
  • 如果二级缓存中不存在,则先将数据加载到二级缓存中,然后再读取二级缓存中的数据。

注意:加载二级缓存的时候需要判断是全量还是增量,如果是增量的话,就从recentlyChangedQueue中加载,如果是全量的话就从registry中加载。

服务同步

服务同步是Server节点之间的数据同步。分为启动时同步,运行时同步。

  • 启动同步

启动同步时,会先遍历Applications中获取的服务信息,并将服务信息注册到registry中。可以参考PeerAwareInstanceRegistryImpl类中的syncUp方法:

public int syncUp() {
       // Copy entire entry from neighboring DS node
       int count = 0;

       for (int i = 0; ((i < serverConfig.getRegistrySyncRetries()) && (count == 0)); i++) {
           if (i > 0) {
               try {
                   Thread.sleep(serverConfig.getRegistrySyncRetryWaitMs());
               } catch (InterruptedException e) {
                   logger.warn("Interrupted during registry transfer..");
                   break;
               }
           }
           Applications apps = eurekaClient.getApplications();
           for (Application app : apps.getRegisteredApplications()) {
               for (InstanceInfo instance : app.getInstances()) {
                   try {
                       if (isRegisterable(instance)) {
                           register(instance, instance.getLeaseInfo().getDurationInSecs(), true);
                           count++;
                       }
                   } catch (Throwable t) {
                       logger.error("During DS init copy", t);
                   }
               }
           }
       }
       return count;
   }

注意这个方法使用类两层for循环,第一次循环时保证自己已经拉取到服务信息,第二层循环是遍历拉取到服务注册信息。

  • 运行时同步

server端当有reigster、renew、cancel请求进来时,会将这些请求封装到一个task中,然后放到一个队列当中,然后经过一系列的处理后,在放到另一个队列中。 可以查看PeerAwareInstanceRegistryImpl类中的BatchWorkerRunnable类,这里就不再贴源码了。

总结

Eureka的原理接介绍到这里,从整体上看似简单,但实现细节相关复杂。得多看几遍源码才能猜透他们的设计思路。

Eureka作为服务的注册与发现,它实际的设计原则是遵循AP原则,也就是“数据的最终一致性”。现在还有好多公司使用zk、nacos来作为服务的注册中心,后续会简单更新一篇关于服务注册中心的对比,这里就不过多阐述。


  • 写作不易,转载请注明出处,喜欢的小伙伴可以关注公众号查看更多喜欢的文章。
  • 联系方式:4272231@163.com QQ:95472323

加载全部内容

相关教程
猜你喜欢
用户评论