Eureka服务注册发现源码流程简析

一： 服务的注册

客户端通过执行InstanceInfoReplicator#run()调用DiscoveryClient#register()发送http请求举行注册
InstanceInfoReplicator 是同于更新同步当前服务到服务端的任务实现
//A task for updating and replicating the local instanceinfo to the remote server.

1. //服务注册
2. boolean register() throws Throwable {
3.         logger.info("DiscoveryClient_{}: registering service...", this.appPathIdentifier);
5.         EurekaHttpResponse httpResponse;
6.         try {
7.             httpResponse = this.eurekaTransport.registrationClient.register(this.instanceInfo);
8.         } catch (Exception var3) {
9.             Exception e = var3;
10.             logger.warn("DiscoveryClient_{} - registration failed {}", new Object[]{this.appPathIdentifier, e.getMessage(), e});
11.             throw e;
12.         }
14.         if (logger.isInfoEnabled()) {
15.             logger.info("DiscoveryClient_{} - registration status: {}", this.appPathIdentifier, httpResponse.getStatusCode());
16.         }
18.         return httpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode();
19.     }
20. //服务续约
21. /**
22. eureka 初始化定时任务，依据设定的心跳时间触发 renew方法，检测服务是否宕机
23. */
24.     boolean renew() {
25.         try {
26.             EurekaHttpResponse<InstanceInfo> httpResponse = this.eurekaTransport.registrationClient.sendHeartBeat(this.instanceInfo.getAppName(), this.instanceInfo.getId(), this.instanceInfo, (InstanceInfo.InstanceStatus)null);
27.             logger.debug("DiscoveryClient_{} - Heartbeat status: {}", this.appPathIdentifier, httpResponse.getStatusCode());
28.             if (httpResponse.getStatusCode() == Status.NOT_FOUND.getStatusCode()) {
29.                 this.REREGISTER_COUNTER.increment();
30.                 logger.info("DiscoveryClient_{} - Re-registering apps/{}", this.appPathIdentifier, this.instanceInfo.getAppName());
31.                 long timestamp = this.instanceInfo.setIsDirtyWithTime();
32.                 boolean success = this.register();
33.                 if (success) {
34.                     this.instanceInfo.unsetIsDirty(timestamp);
35.                 }
37.                 return success;
38.             } else {
39.                 return httpResponse.getStatusCode() == Status.OK.getStatusCode();
40.             }
41.         } catch (Throwable var5) {
42.             Throwable e = var5;
43.             logger.error("DiscoveryClient_{} - was unable to send heartbeat!", this.appPathIdentifier, e);
44.             return false;
45.         }
46.     }

复制代码

服务端服务注册接受和存储

1. //eureka 客户端会通过此方法注册保存到eureka server 的内存中
2. public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
3.         this.read.lock();
5.         try {
6.             Map<String, Lease<InstanceInfo>> gMap = (Map)this.registry.get(registrant.getAppName());
7.             EurekaMonitors.REGISTER.increment(isReplication);
8.             if (gMap == null) {
9.                 ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap();
10.                 gMap = (Map)this.registry.putIfAbsent(registrant.getAppName(), gNewMap);
11.                 if (gMap == null) {
12.                     gMap = gNewMap;
13.                 }
14.             }
16.             Lease<InstanceInfo> existingLease = (Lease)((Map)gMap).get(registrant.getId());
17.             if (existingLease != null && existingLease.getHolder() != null) {
18.                 Long existingLastDirtyTimestamp = ((InstanceInfo)existingLease.getHolder()).getLastDirtyTimestamp();
19.                 Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
20.                 logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
21.                 if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
22.                     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);
23.                     logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");
24.                     registrant = (InstanceInfo)existingLease.getHolder();
25.                 }
26.             } else {
27.                 synchronized(this.lock) {
28.                     if (this.expectedNumberOfClientsSendingRenews > 0) {
29.                         ++this.expectedNumberOfClientsSendingRenews;
30.                         this.updateRenewsPerMinThreshold();
31.                     }
32.                 }
34.                 logger.debug("No previous lease information found; it is new registration");
35.             }
37.             Lease<InstanceInfo> lease = new Lease(registrant, leaseDuration);
38.             if (existingLease != null) {
39.                 lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
40.             }
42.             ((Map)gMap).put(registrant.getId(), lease);
43.             this.recentRegisteredQueue.add(new Pair(System.currentTimeMillis(), registrant.getAppName() + "(" + registrant.getId() + ")"));
44.             if (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) {
45.                 logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the overrides", registrant.getOverriddenStatus(), registrant.getId());
46.                 if (!this.overriddenInstanceStatusMap.containsKey(registrant.getId())) {
47.                     logger.info("Not found overridden id {} and hence adding it", registrant.getId());
48.                     this.overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
49.                 }
50.             }
52.             InstanceInfo.InstanceStatus overriddenStatusFromMap = (InstanceInfo.InstanceStatus)this.overriddenInstanceStatusMap.get(registrant.getId());
53.             if (overriddenStatusFromMap != null) {
54.                 logger.info("Storing overridden status {} from map", overriddenStatusFromMap);
55.                 registrant.setOverriddenStatus(overriddenStatusFromMap);
56.             }
58.             InstanceInfo.InstanceStatus overriddenInstanceStatus = this.getOverriddenInstanceStatus(registrant, existingLease, isReplication);
59.             registrant.setStatusWithoutDirty(overriddenInstanceStatus);
60.             if (InstanceStatus.UP.equals(registrant.getStatus())) {
61.                 lease.serviceUp();
62.             }
64.             registrant.setActionType(ActionType.ADDED);
65.             this.recentlyChangedQueue.add(new RecentlyChangedItem(lease));
66.             registrant.setLastUpdatedTimestamp();
67.             this.invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
68.             logger.info("Registered instance {}/{} with status {} (replication={})", new Object[]{registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication});
69.         } finally {
70.             this.read.unlock();
71.         }
73.     }

复制代码

如图，registry保存有将注册、已注册到server 的eureka客户端 instance信息；
将要注册到注册表registry中的instance，会创建一个map结构保存

二： 服务的发现

依据上图展示，服务注册采取的是客户端建立DiscoveryClient建立http请求，同理使用此DiscoveryClient实例通过请求完成服务的发现，不再赘述；
重点将放入服务发现的缓存和调用

1.

1. //其他发现实现 DiscoveryClient#getAndUpdateDelta 包括更新等具体操作不作继续深入讨论
2.     /**
3.      * Gets the full registry information from the eureka server and stores it locally.
4.      * When applying the full registry, the following flow is observed:
5.      *
6.      * if (update generation have not advanced (due to another thread))
7.      *   atomically set the registry to the new registry
8.      * fi
9.      *
10.      * @return the full registry information.
11.      * @throws Throwable
12.      *             on error.
13.      */
14.     private void getAndStoreFullRegistry() throws Throwable {
15.         long currentUpdateGeneration = fetchRegistryGeneration.get();
17.         logger.info("Getting all instance registry info from the eureka server");
19.         Applications apps = null;
20.         EurekaHttpResponse<Applications> httpResponse = clientConfig.getRegistryRefreshSingleVipAddress() == null
21.                 ? eurekaTransport.queryClient.getApplications(remoteRegionsRef.get())
22.                 : eurekaTransport.queryClient.getVip(clientConfig.getRegistryRefreshSingleVipAddress(), remoteRegionsRef.get());
23.         if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
24.             apps = httpResponse.getEntity();
25.         }
26.         logger.info("The response status is {}", httpResponse.getStatusCode());
28.         if (apps == null) {
29.             logger.error("The application is null for some reason. Not storing this information");
30.         } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
31.     //存入本地服务缓存，即客户端将server端的服务注册信息缓存了一份，存在后续的缓存更新机制不做深入
32.     //this.filterAndShuffle(apps) 目的是处理 UP 状态的实例以及 打乱保证随机性
33.     // Shuffling helps in randomizing the applications list there by avoiding the same instances receiving traffic during start ups.
34.             localRegionApps.set(this.filterAndShuffle(apps));
35.             logger.debug("Got full registry with apps hashcode {}", apps.getAppsHashCode());
36.         } else {
37.             logger.warn("Not updating applications as another thread is updating it already");
38.         }
39.     }

复制代码

2. 调用

一： 使用DiscoveryClient获取实例的信息，再构建http请求
二： 使用组件fegin完成服务转发

1. @FeignClient(value = "eurekaclient")
2. public interface ApiService {
4.    @RequestMapping(value = "/index",method = RequestMethod.GET)
5.    String index();
6. }
8. /**
9. 等价于 new httpclient(eurekaClient) => 发送 /index 接口并接受到response
10. */

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。