中国教育网站官网,网站开发需求分析报告,兰州解封最新消息,oracle网站开发看过之前文章的朋友们#xff0c;相信已经对Eureka的运行机制已经有了一定的了解。为了更深入的理解它的运作和配置#xff0c;下面我们结合源码来分别看看服务端和客户端的通信行为是如何实现的。另外写这篇文章#xff0c;还有一个目的#xff0c;还是希望鼓励大家能够学…看过之前文章的朋友们相信已经对Eureka的运行机制已经有了一定的了解。为了更深入的理解它的运作和配置下面我们结合源码来分别看看服务端和客户端的通信行为是如何实现的。另外写这篇文章还有一个目的还是希望鼓励大家能够学会学习和研究的方法由于目前Spring Cloud的中文资料并不多并不是大部分的问题都能找到现成的答案所以其实很多问题给出一个科学而慎重的解答也都是花费研究者不少精力的。
在看具体源码前我们先回顾一下之前我们所实现的内容从而找一个合适的切入口去分析。首先服务注册中心、服务提供者、服务消费者这三个主要元素来说后两者也就是Eureka客户端在整个运行机制中是大部分通信行为的主动发起者而注册中心主要是处理请求的接收者。所以我们可以从Eureka的客户端作为入口看看它是如何完成这些主动通信行为的。
我们在将一个普通的Spring Boot应用注册到Eureka Server中或是从Eureka Server中获取服务列表时主要就做了两件事
在应用主类中配置了EnableDiscoveryClient注解在application.properties中用eureka.client.serviceUrl.defaultZone参数指定了服务注册中心的位置
顺着上面的线索我们先查看EnableDiscoveryClient的源码如下
/** * Annotation to enable a DiscoveryClient implementation. * author Spencer Gibb */Target(ElementType.TYPE)Retention(RetentionPolicy.RUNTIME)DocumentedInheritedImport(EnableDiscoveryClientImportSelector.class)public interface EnableDiscoveryClient {}从该注解的注释我们可以知道该注解用来开启DiscoveryClient的实例。通过搜索DiscoveryClient我们可以发现有一个类和一个接口。通过梳理可以得到如下图的关系 其中左边的org.springframework.cloud.client.discovery.DiscoveryClient是Spring Cloud的接口它定义了用来发现服务的常用抽象方法而org.springframework.cloud.netflix.eureka.EurekaDiscoveryClient是对该接口的实现从命名来就可以判断它实现的是对Eureka发现服务的封装。所以EurekaDiscoveryClient依赖了Eureka的com.netflix.discovery.EurekaClient接口EurekaClient继承了LookupService接口他们都是Netflix开源包中的内容它主要定义了针对Eureka的发现服务的抽象方法而真正实现发现服务的则是Netflix包中的com.netflix.discovery.DiscoveryClient类。
那么我们就看看来详细看看DiscoveryClient类。先解读一下该类头部的注释有个总体的了解注释的大致内容如下
这个类用于帮助与Eureka Server互相协作。Eureka Client负责了下面的任务- 向Eureka Server注册服务实例- 向Eureka Server为租约续期- 当服务关闭期间向Eureka Server取消租约- 查询Eureka Server中的服务实例列表Eureka Client还需要配置一个Eureka Server的URL列表。在具体研究Eureka Client具体负责的任务之前我们先看看对Eureka Server的URL列表配置在哪里。根据我们配置的属性名eureka.client.serviceUrl.defaultZone通过serviceUrl我们找到该属性相关的加载属性但是在SR5版本中它们都被Deprecated标注了并在注视中可以看到link到了替代类com.netflix.discovery.endpoint.EndpointUtils我们可以在该类中找到下面这个函数
public static MapString, ListString getServiceUrlsMapFromConfig( EurekaClientConfig clientConfig, String instanceZone, boolean preferSameZone) { MapString, ListString orderedUrls new LinkedHashMap(); String region getRegion(clientConfig); String[] availZones clientConfig.getAvailabilityZones(clientConfig.getRegion()); if (availZones null || availZones.length 0) { availZones new String[1]; availZones[0] DEFAULT_ZONE; } …… int myZoneOffset getZoneOffset(instanceZone, preferSameZone, availZones); String zone availZones[myZoneOffset]; ListString serviceUrls clientConfig.getEurekaServerServiceUrls(zone); if (serviceUrls ! null) { orderedUrls.put(zone, serviceUrls); } …… return orderedUrls;}Region、Zone
在上面的函数中我们可以发现客户端依次加载了两个内容第一个是Region第二个是Zone从其加载逻上我们可以判断他们之间的关系
通过getRegion函数我们可以看到它从配置中读取了一个Region返回所以一个微服务应用只可以属于一个Region如果不特别配置就默认为default。若我们要自己设置可以通过eureka.client.region属性来定义。
public static String getRegion(EurekaClientConfig clientConfig) { String region clientConfig.getRegion(); if (region null) { region DEFAULT_REGION; } region region.trim().toLowerCase(); return region;}通过getAvailabilityZones函数我们可以知道当我们没有特别为Region配置Zone的时候将默认采用defaultZone这也是我们之前配置参数eureka.client.serviceUrl.defaultZone的由来。若要为应用指定Zone我们可以通过eureka.client.availability-zones属性来进行设置。从该函数的return内容我们可以Zone是可以有多个的并且通过逗号分隔来配置。由此我们可以判断Region与Zone是一对多的关系。
public String[] getAvailabilityZones(String region) { String value this.availabilityZones.get(region); if (value null) { value DEFAULT_ZONE; } return value.split(,);}ServiceUrls
在获取了Region和Zone信息之后才开始真正加载Eureka Server的具体地址。它根据传入的参数按一定算法确定加载位于哪一个Zone配置的serviceUrls。
int myZoneOffset getZoneOffset(instanceZone, preferSameZone, availZones);String zone availZones[myZoneOffset];ListString serviceUrls clientConfig.getEurekaServerServiceUrls(zone);具体获取serviceUrls的实现我们可以详细查看getEurekaServerServiceUrls函数的具体实现类EurekaClientConfigBean该类是EurekaClientConfig和EurekaConstants接口的实现用来加载配置文件中的内容这里有非常多有用的信息这里我们先说一下此处我们关心的关于defaultZone的信息。通过搜索defaultZone我们可以很容易的找到下面这个函数它具体实现了如何解析该参数的过程通过此内容我们就可以知道eureka.client.serviceUrl.defaultZone属性可以配置多个并且需要通过逗号分隔。
public ListString getEurekaServerServiceUrls(String myZone) { String serviceUrls this.serviceUrl.get(myZone); if (serviceUrls null || serviceUrls.isEmpty()) { serviceUrls this.serviceUrl.get(DEFAULT_ZONE); } if (!StringUtils.isEmpty(serviceUrls)) { final String[] serviceUrlsSplit StringUtils.commaDelimitedListToStringArray(serviceUrls); ListString eurekaServiceUrls new ArrayList(serviceUrlsSplit.length); for (String eurekaServiceUrl : serviceUrlsSplit) { if (!endsWithSlash(eurekaServiceUrl)) { eurekaServiceUrl /; } eurekaServiceUrls.add(eurekaServiceUrl); } return eurekaServiceUrls; } return new ArrayList();}当客户端在服务列表中选择实例进行访问时对于Zone和Region遵循这样的规则优先访问同自己一个Zone中的实例其次才访问其他Zone中的实例。通过Region和Zone的两层级别定义配合实际部署的物理结构我们就可以有效的设计出区域性故障的容错集群。
服务注册
在理解了多个服务注册中心信息的加载后我们再回头看看DiscoveryClient类是如何实现“服务注册”行为的通过查看它的构造类可以找到它调用了下面这个函数
private void initScheduledTasks() { ... if (clientConfig.shouldRegisterWithEureka()) { ... // InstanceInfo replicator instanceInfoReplicator new InstanceInfoReplicator( this, instanceInfo, clientConfig.getInstanceInfoReplicationIntervalSeconds(), 2); // burstSize ... instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds()); } else { logger.info(Not registering with Eureka server per configuration); }}在上面的函数中我们可以看到关键的判断依据if (clientConfig.shouldRegisterWithEureka())。在该分支内创建了一个InstanceInfoReplicator类的实例它会执行一个定时任务查看该类的run()函数了解该任务做了什么工作
public void run() { try { discoveryClient.refreshInstanceInfo(); Long dirtyTimestamp instanceInfo.isDirtyWithTime(); if (dirtyTimestamp ! null) { discoveryClient.register(); instanceInfo.unsetIsDirty(dirtyTimestamp); } } catch (Throwable t) { logger.warn(There was a problem with the instance info replicator, t); } finally { Future next scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS); scheduledPeriodicRef.set(next); }}相信大家都发现了discoveryClient.register();这一行真正触发调用注册的地方就在这里。继续查看register()的实现内容如下
boolean register() throws Throwable { logger.info(PREFIX appPathIdentifier : registering service...); EurekaHttpResponseVoid httpResponse; try { httpResponse eurekaTransport.registrationClient.register(instanceInfo); } catch (Exception e) { logger.warn({} - registration failed {}, PREFIX appPathIdentifier, e.getMessage(), e); throw e; } if (logger.isInfoEnabled()) { logger.info({} - registration status: {}, PREFIX appPathIdentifier, httpResponse.getStatusCode()); } return httpResponse.getStatusCode() 204;}通过属性命名大家基本也能猜出来注册操作也是通过REST请求的方式进行的。同时这里我们也能看到发起注册请求的时候传入了一个com.netflix.appinfo.InstanceInfo对象该对象就是注册时候客户端给服务端的服务的元数据。
服务获取与服务续约
顺着上面的思路我们继续来看DiscoveryClient的initScheduledTasks函数不难发现在其中还有两个定时任务分别是“服务获取”和“服务续约”
private void initScheduledTasks() { if (clientConfig.shouldFetchRegistry()) { // registry cache refresh timer int registryFetchIntervalSeconds clientConfig.getRegistryFetchIntervalSeconds(); int expBackOffBound clientConfig.getCacheRefreshExecutorExponentialBackOffBound(); scheduler.schedule( new TimedSupervisorTask( cacheRefresh, scheduler, cacheRefreshExecutor, registryFetchIntervalSeconds, TimeUnit.SECONDS, expBackOffBound, new CacheRefreshThread() ), registryFetchIntervalSeconds, TimeUnit.SECONDS); } if (clientConfig.shouldRegisterWithEureka()) { int renewalIntervalInSecs instanceInfo.getLeaseInfo().getRenewalIntervalInSecs(); int expBackOffBound clientConfig.getHeartbeatExecutorExponentialBackOffBound(); logger.info(Starting heartbeat executor: renew interval is: renewalIntervalInSecs); // Heartbeat timer scheduler.schedule( new TimedSupervisorTask( heartbeat, scheduler, heartbeatExecutor, renewalIntervalInSecs, TimeUnit.SECONDS, expBackOffBound, new HeartbeatThread() ), renewalIntervalInSecs, TimeUnit.SECONDS); // InstanceInfo replicator …… }}从源码中我们就可以发现“服务获取”相对于“服务续约”更为独立“服务续约”与“服务注册”在同一个if逻辑中这个不难理解服务注册到Eureka Server后自然需要一个心跳去续约防止被剔除所以他们肯定是成对出现的。从源码中我们可以清楚看到了对于服务续约相关的时间控制参数
eureka.instance.lease-renewal-interval-in-seconds30eureka.instance.lease-expiration-duration-in-seconds90而“服务获取”的逻辑在独立的一个if判断中其判断依据就是我们之前所提到的eureka.client.fetch-registrytrue参数它默认是为true的大部分情况下我们不需要关心。为了定期的更新客户端的服务清单以保证服务访问的正确性“服务获取”的请求不会只限于服务启动而是一个定时执行的任务从源码中我们可以看到任务运行中的registryFetchIntervalSeconds参数对应eureka.client.registry-fetch-interval-seconds30配置参数它默认为30秒。
继续循序渐进的向下深入我们就能分别发现实现“服务获取”和“服务续约”的具体方法其中“服务续约”的实现较为简单直接以REST请求的方式进行续约
boolean renew() { EurekaHttpResponseInstanceInfo httpResponse; try { httpResponse eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null); logger.debug({} - Heartbeat status: {}, PREFIX appPathIdentifier, httpResponse.getStatusCode()); if (httpResponse.getStatusCode() 404) { REREGISTER_COUNTER.increment(); logger.info({} - Re-registering apps/{}, PREFIX appPathIdentifier, instanceInfo.getAppName()); return register(); } return httpResponse.getStatusCode() 200; } catch (Throwable e) { logger.error({} - was unable to send heartbeat!, PREFIX appPathIdentifier, e); return false; }}而“服务获取”则相对复杂一些会根据是否第一次获取发起不同的REST请求和相应的处理具体的实现逻辑还是跟之前类似有兴趣的读者可以继续查看服务客户端的其他具体内容了解更多细节。
服务注册中心处理
通过上面的源码分析可以看到所有的交互都是通过REST的请求来发起的。下面我们来看看服务注册中心对这些请求的处理。Eureka Server对于各类REST请求的定义都位于com.netflix.eureka.resources包下。
以“服务注册”请求为例
POSTConsumes({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 ... // 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}在对注册信息进行了一大堆校验之后会调用org.springframework.cloud.netflix.eureka.server.InstanceRegistry对象中的register(InstanceInfo info, int leaseDuration, boolean isReplication)函数来进行服务注册
public void register(InstanceInfo info, int leaseDuration, boolean isReplication) { if (log.isDebugEnabled()) { log.debug(register info.getAppName() , vip info.getVIPAddress() , leaseDuration leaseDuration , isReplication isReplication); } this.ctxt.publishEvent(new EurekaInstanceRegisteredEvent(this, info, leaseDuration, isReplication)); super.register(info, leaseDuration, isReplication);}在注册函数中先调用publishEvent函数将该新服务注册的事件传播出去然后调用com.netflix.eureka.registry.AbstractInstanceRegistry父类中的注册实现将InstanceInfo中的元数据信息存储在一个ConcurrentHashMapString, MapString, LeaseInstanceInfo对象中它是一个两层Map结构第一层的key存储服务名InstanceInfo中的appName属性第二层的key存储实例名InstanceInfo中的instanceId属性。
服务端的请求接收都非常类似对于其他的服务端处理这里就不再展开读者可以根据上面的脉络来自己查看其内容这里包含很多细节内容来帮助和加深理解。
