Hike News
Hike News

Eureka服务注册发现

java 本文总阅读量次

注册发现流程

通常服务注册发现都分为服务提供者（provider）、服务消费者（consumer）、注册中心（register）三个部分。

一个服务可以是服务提供者可以是服务消费者，也可以即是提供者也是消费者

其中provider和consumer作为客户端client，注册中心作为服务端server，都可以同时存在多个实例。

服务提供者把自身的实例信息（比如IP、PORT、状态等）在注册中心做登记，服务消费者通过注册中心获取服务提供者的信息然后发起调用。

常用的注册中心有Zookeeper、ETCD、Eureka、Consul、Nacos。

服务注册

首先我们会在client配置eureka server的地址，如：

1
2
eureka.client.service-url.defaultZone=
http://localhost:8081/eureka/,http://localhost:8082/eureka/,http://localhost:8083/eureka/

客户端会尝试向配置的eureka server发起注册请求。

入口：com.netflix.discovery.DiscoveryClient#register

1
2
3
4
5
6
7
8
9
10
11
12
13
14
boolean register() throws Throwable {
        logger.info(PREFIX + appPathIdentifier + ": registering service...");
        EurekaHttpResponse<Void> 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;
    }

核心代码：com.netflix.discovery.shared.transport.decorator.RetryableEurekaHttpClient#execute

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
protected <R> EurekaHttpResponse<R> execute(RequestExecutor<R> requestExecutor) {
       List<EurekaEndpoint> candidateHosts = null;
       int endpointIdx = 0;
       for (int retry = 0; retry < numberOfRetries; retry++) {
           //currentHttpClient是上面配置的三个eureka server中的其中一个
           //如果eureka server能用，就会一直请求同一个eureka server
           EurekaHttpClient currentHttpClient = delegate.get();
           EurekaEndpoint currentEndpoint = null;
           if (currentHttpClient == null) {
               if (candidateHosts == null) {
                   //获取eureka server列表
                   candidateHosts = getHostCandidates();
                   if (candidateHosts.isEmpty()) {
                       throw new TransportException("There is no known eureka server; cluster server list is empty");
                   }
               }
               if (endpointIdx >= candidateHosts.size()) {
                   throw new TransportException("Cannot execute request on any known server");
               }

               currentEndpoint = candidateHosts.get(endpointIdx++);
               currentHttpClient = clientFactory.newClient(currentEndpoint);
           }

           try {
               EurekaHttpResponse<R> response = requestExecutor.execute(currentHttpClient);
               if (serverStatusEvaluator.accept(response.getStatusCode(), requestExecutor.getRequestType())) {
                   delegate.set(currentHttpClient);
                   if (retry > 0) {
                       logger.info("Request execution succeeded on retry #{}", retry);
                   }
                   return response;
               }
               logger.warn("Request execution failure with status code {}; retrying on another server if available", response.getStatusCode());
           } catch (Exception e) {
               logger.warn("Request execution failed with message: {}", e.getMessage());  // just log message as the underlying client should log the stacktrace
           }

           //如果请求则把eureka server置为空，下次重试时会换一个eureka server
           delegate.compareAndSet(currentHttpClient, null);
           if (currentEndpoint != null) {
               quarantineSet.add(currentEndpoint);
           }
       }
       throw new TransportException("Retry limit reached; giving up on completing the request");
   }

Eureka server获取

我们实际注册的eureka server和配置的eureka server顺序是不一样的，比如我们配置的顺序是http://localhost:8081/eureka,http://localhost:8082/eureka,http://localhost:8083/eureka，原先以为会先注册到http://localhost:8081/eureka，但实际是先尝试注册到http://localhost:8083/eureka。

因为获取列表的时候做了一次randomize，如果是三个server实例，会交互1和3的位置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public static <T extends EurekaEndpoint> List<T> randomize(List<T> list) {
        List<T> randomList = new ArrayList<>(list);
        if (randomList.size() < 2) {
            return randomList;
        }
        Random random = new Random(LOCAL_IPV4_ADDRESS.hashCode());
        int last = randomList.size() - 1;
        for (int i = 0; i < last; i++) {
            int pos = random.nextInt(randomList.size() - i);
            if (pos != i) {
                Collections.swap(randomList, i, pos);
            }
        }
        return randomList;
    }

客户端发起注册请求

源码：com.netflix.discovery.shared.transport.jersey.AbstractJerseyEurekaHttpClient#register

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public EurekaHttpResponse<Void> register(InstanceInfo info) {
        String urlPath = "apps/" + info.getAppName();
        ClientResponse response = null;
        try {
            Builder resourceBuilder = jerseyClient.resource(serviceUrl).path(urlPath).getRequestBuilder();
            addExtraHeaders(resourceBuilder);
            response = resourceBuilder
                    .header("Accept-Encoding", "gzip")
                    .type(MediaType.APPLICATION_JSON_TYPE)
                    .accept(MediaType.APPLICATION_JSON)
                    .post(ClientResponse.class, info);
            return anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
        } finally {
            if (logger.isDebugEnabled()) {
                logger.debug("Jersey HTTP POST {}/{} with instance {}; statusCode={}", serviceUrl, urlPath, info.getId(),
                        response == null ? "N/A" : response.getStatus());
            }
            if (response != null) {
                response.close();
            }
        }
    }

服务端接受注册请求

入口：com.netflix.eureka.resources.ApplicationResource#addInstance

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@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.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
   }

核心代码：com.netflix.eureka.registry.AbstractInstanceRegistry

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
    try {
        // 上只读锁
        read.lock();
        // 从本地MAP里面获取当前实例的信息。
        Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
        // 增加注册次数到监控信息里面去。
        REGISTER.increment(isReplication);
        if (gMap == null) {
            // 如果第一次进来，那么gMap为空，则创建一个ConcurrentHashMap放入到registry里面去
            final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
            // putIfAbsent方法主要是在向ConcurrentHashMap中添加键—值对的时候，它会先判断该键值对是否已经存在。
            // 如果不存在（新的entry），那么会向map中添加该键值对，并返回null。
            // 如果已经存在，那么不会覆盖已有的值，直接返回已经存在的值。
            gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
            if (gMap == null) {
                // 表明map中确实不存在，则设置gMap为最新创建的那个
                gMap = gNewMap;
            }
        }
        // 从MAP中查询已经存在的Lease信息 （比如第二次来）
        Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
        // 当Lease的对象不为空时。
        if (existingLease != null && (existingLease.getHolder() != null)) {
            // 当instance已经存在是，和客户端的instance的信息做比较，时间最新的那个，为有效instance信息
            Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp(); // server
            Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();   // client
            logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
            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 {
            // 这里只有当existinglease不存在时，才会进来。 像那种恢复心跳，信息过期的，都不会进入这里。
            //  Eureka-Server的自我保护机制做的操作，为每分钟最大续约数+2 ，同时重新计算每分钟最小续约数
            synchronized (lock) {
                if (this.expectedNumberOfRenewsPerMin > 0) {
                    // Since the client wants to cancel it, reduce the threshold
                    // (1
                    // for 30 seconds, 2 for a minute)
                    this.expectedNumberOfRenewsPerMin = this.expectedNumberOfRenewsPerMin + 2;
                    this.numberOfRenewsPerMinThreshold =
                            (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());
                }
            }
            logger.debug("No previous lease information found; it is new registration");
        }
        // 构建一个最新的Lease信息
        Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
        if (existingLease != null) {
            // 当原来存在Lease的信息时，设置他的serviceUpTimestamp, 保证服务开启的时间一直是第一次的那个
            lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
        }
        // 放入本地Map中
        gMap.put(registrant.getId(), lease);
        // 添加到最近的注册队列里面去，以时间戳作为Key， 名称作为value，主要是为了运维界面的统计数据。
        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
        // 分析instanceStatus
        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
        // 得到instanceStatus，判断是否是UP状态，
        if (InstanceStatus.UP.equals(registrant.getStatus())) {
            lease.serviceUp();
        }
        // 设置注册类型为添加
        registrant.setActionType(ActionType.ADDED);
        // 租约变更记录队列，记录了实例的每次变化， 用于注册信息的增量获取、
        recentlyChangedQueue.add(new RecentlyChangedItem(lease));
        registrant.setLastUpdatedTimestamp();
        // 清理缓存 ，传入的参数为key
        invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
        logger.info("Registered instance {}/{} with status {} (replication={})",
                registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);
    } finally {
        read.unlock();
    }
}

到这里，服务注册就算完啦。

服务发现

Eureka server缓存

eureka server默认情况下有三个地方存储了服务信息，分别是com.netflix.eureka.registry.AbstractInstanceRegistry#registry、com.netflix.eureka.registry.ResponseCacheImpl#readWriteCacheMap、com.netflix.eureka.registry.ResponseCacheImpl#readOnlyCacheMap。

registry

数据结构ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>，外面Map的key是AppName，就是注册在eureka上的服务；里面Map的key是IP，value是服务的具体实例。

所有的服务都是注册到registry上面的，eureka server提供的UI界面查询到的服务信息也是直接从registry中获取的。

入口：org.springframework.cloud.netflix.eureka.server.EurekaController#status

核心代码：com.netflix.eureka.registry.AbstractInstanceRegistry#getApplicationsFromMultipleRegions

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
public Applications getApplicationsFromMultipleRegions(String[] remoteRegions) {

        boolean includeRemoteRegion = null != remoteRegions && remoteRegions.length != 0;

        logger.debug("Fetching applications registry with remote regions: {}, Regions argument {}",
                includeRemoteRegion, Arrays.toString(remoteRegions));

        if (includeRemoteRegion) {
            GET_ALL_WITH_REMOTE_REGIONS_CACHE_MISS.increment();
        } else {
            GET_ALL_CACHE_MISS.increment();
        }
        Applications apps = new Applications();
        apps.setVersion(1L);
        for (Entry<String, Map<String, Lease<InstanceInfo>>> entry : registry.entrySet()) {
            Application app = null;

            if (entry.getValue() != null) {
                for (Entry<String, Lease<InstanceInfo>> stringLeaseEntry : entry.getValue().entrySet()) {
                    Lease<InstanceInfo> lease = stringLeaseEntry.getValue();
                    if (app == null) {
                        app = new Application(lease.getHolder().getAppName());
                    }
                    app.addInstance(decorateInstanceInfo(lease));
                }
            }
            if (app != null) {
                apps.addApplication(app);
            }
        }
        if (includeRemoteRegion) {
            for (String remoteRegion : remoteRegions) {
                RemoteRegionRegistry remoteRegistry = regionNameVSRemoteRegistry.get(remoteRegion);
                if (null != remoteRegistry) {
                    Applications remoteApps = remoteRegistry.getApplications();
                    for (Application application : remoteApps.getRegisteredApplications()) {
                        if (shouldFetchFromRemoteRegistry(application.getName(), remoteRegion)) {
                            logger.info("Application {}  fetched from the remote region {}",
                                    application.getName(), remoteRegion);

                            Application appInstanceTillNow = apps.getRegisteredApplications(application.getName());
                            if (appInstanceTillNow == null) {
                                appInstanceTillNow = new Application(application.getName());
                                apps.addApplication(appInstanceTillNow);
                            }
                            for (InstanceInfo instanceInfo : application.getInstances()) {
                                appInstanceTillNow.addInstance(instanceInfo);
                            }
                        } else {
                            logger.debug("Application {} not fetched from the remote region {} as there exists a "
                                            + "whitelist and this app is not in the whitelist.",
                                    application.getName(), remoteRegion);
                        }
                    }
                } else {
                    logger.warn("No remote registry available for the remote region {}", remoteRegion);
                }
            }
        }
        apps.setAppsHashCode(apps.getReconcileHashCode());
        return apps;
    }

readWriteCacheMap

数据结构LoadingCache<Key, Value>。

LoadingCache是谷歌guava提供的一个缓存实现，核心代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
this.readWriteCacheMap =
               CacheBuilder.newBuilder().initialCapacity(1000)
   					//缓存失效时间，可以配置，默认180秒
                       .expireAfterWrite(serverConfig.getResponseCacheAutoExpirationInSeconds(), TimeUnit.SECONDS)
                       .removalListener(new RemovalListener<Key, Value>() {
                           @Override
                           public void onRemoval(RemovalNotification<Key, Value> notification) {
                               Key removedKey = notification.getKey();
                               if (removedKey.hasRegions()) {
                                   Key cloneWithNoRegions = removedKey.cloneWithoutRegions();
                                   regionSpecificKeys.remove(cloneWithNoRegions, removedKey);
                               }
                           }
                       })
   				//如果key对应的value为空，调用该方法获取value并放到缓存中，该方法是线程安全的
   				//这里是去registry中获取
                       .build(new CacheLoader<Key, Value>() {
                           @Override
                           public Value load(Key key) throws Exception {
                               if (key.hasRegions()) {
                                   Key cloneWithNoRegions = key.cloneWithoutRegions();
                                   regionSpecificKeys.put(cloneWithNoRegions, key);
                               }
                               Value value = generatePayload(key);
                               return value;
                           }
                       });

   /*
    * Generate pay load for the given key.
    */
   private Value generatePayload(Key key) {
       Stopwatch tracer = null;
       try {
           String payload;
           switch (key.getEntityType()) {
               case Application:
                   boolean isRemoteRegionRequested = key.hasRegions();
    
                   if (ALL_APPS.equals(key.getName())) {
                       if (isRemoteRegionRequested) {
                           tracer = serializeAllAppsWithRemoteRegionTimer.start();
                           payload = getPayLoad(key, registry.getApplicationsFromMultipleRegions(key.getRegions()));
                       } else {
                           tracer = serializeAllAppsTimer.start();
                           payload = getPayLoad(key, registry.getApplications());
                       }
                   } else if (ALL_APPS_DELTA.equals(key.getName())) {
                       if (isRemoteRegionRequested) {
                           tracer = serializeDeltaAppsWithRemoteRegionTimer.start();
                           versionDeltaWithRegions.incrementAndGet();
                           versionDeltaWithRegionsLegacy.incrementAndGet();
                           payload = getPayLoad(key, registry.getApplicationDeltasFromMultipleRegions(key.getRegions()));
                       } else {
                           tracer = serializeDeltaAppsTimer.start();
                           versionDelta.incrementAndGet();
                           versionDeltaLegacy.incrementAndGet();
                           payload = getPayLoad(key, registry.getApplicationDeltas());
                       }
                   } else {
                       tracer = serializeOneApptimer.start();
                       payload = getPayLoad(key, registry.getApplication(key.getName()));
                   }
                   break;
               case VIP:
               case SVIP:
                   tracer = serializeViptimer.start();
                   payload = getPayLoad(key, getApplicationsForVip(key, registry));
                   break;
               default:
                   logger.error("Unidentified entity type: " + key.getEntityType() + " found in the cache key.");
                   payload = "";
                   break;
           }
           return new Value(payload);
       } finally {
           if (tracer != null) {
               tracer.stop();
           }
       }
   }

缓存除了到失效时间后会失效以外，每次对registry进行写操作的时候也会调用com.netflix.eureka.registry.ResponseCacheImpl#invalidate删除缓存。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
public void invalidate(String appName, @Nullable String vipAddress, @Nullable String secureVipAddress) {
      for (Key.KeyType type : Key.KeyType.values()) {
          for (Version v : Version.values()) {
              invalidate(
                      new Key(Key.EntityType.Application, appName, type, v, EurekaAccept.full),
                      new Key(Key.EntityType.Application, appName, type, v, EurekaAccept.compact),
                      new Key(Key.EntityType.Application, ALL_APPS, type, v, EurekaAccept.full),
                      new Key(Key.EntityType.Application, ALL_APPS, type, v, EurekaAccept.compact),
                      new Key(Key.EntityType.Application, ALL_APPS_DELTA, type, v, EurekaAccept.full),
                      new Key(Key.EntityType.Application, ALL_APPS_DELTA, type, v, EurekaAccept.compact)
              );
              if (null != vipAddress) {
                  invalidate(new Key(Key.EntityType.VIP, vipAddress, type, v, EurekaAccept.full));
              }
              if (null != secureVipAddress) {
                  invalidate(new Key(Key.EntityType.SVIP, secureVipAddress, type, v, EurekaAccept.full));
              }
          }
      }
  }
  public void invalidate(Key... keys) {
      for (Key key : keys) {
          logger.debug("Invalidating the response cache key : {} {} {} {}, {}",
                  key.getEntityType(), key.getName(), key.getVersion(), key.getType(), key.getEurekaAccept());
    
          readWriteCacheMap.invalidate(key);
          Collection<Key> keysWithRegions = regionSpecificKeys.get(key);
          if (null != keysWithRegions && !keysWithRegions.isEmpty()) {
              for (Key keysWithRegion : keysWithRegions) {
                  logger.debug("Invalidating the response cache key : {} {} {} {} {}",
                          key.getEntityType(), key.getName(), key.getVersion(), key.getType(), key.getEurekaAccept());
                  readWriteCacheMap.invalidate(keysWithRegion);
              }
          }
      }
  }

由此可以看出，readWriteCacheMap的数据和registry的数据基本是一致的。

readOnlyCacheMap

数据结构ConcurrentMap<Key, Value>。

可以用useReadOnlyResponseCache配置控制是否启用readOnlyCacheMap，默认为true启用readOnlyCacheMap。

初始化com.netflix.eureka.registry.ResponseCacheImpl的时候会判断useReadOnlyResponseCache,启用的话会开一个更新readOnlyCacheMap的定时任务：

1
2
3
4
5
6
if (shouldUseReadOnlyResponseCache) {
            timer.schedule(getCacheUpdateTask(),
                    new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
                            + responseCacheUpdateIntervalMs),
                    responseCacheUpdateIntervalMs);
        }

responseCacheUpdateIntervalMs是readOnlyCacheMap从readWriteCacheMap同步缓存的时间，默认30s。

核心代码：com.netflix.eureka.registry.ResponseCacheImpl#getCacheUpdateTask

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
private TimerTask getCacheUpdateTask() {
       return new TimerTask() {
           @Override
           public void run() {
               logger.debug("Updating the client cache from response cache");
               for (Key key : readOnlyCacheMap.keySet()) {
                   if (logger.isDebugEnabled()) {
                       Object[] args = {key.getEntityType(), key.getName(), key.getVersion(), key.getType()};
                       logger.debug("Updating the client cache from response cache for key : {} {} {} {}", args);
                   }
                   try {
                       CurrentRequestVersion.set(key.getVersion());
                       Value cacheValue = readWriteCacheMap.get(key);
                       Value currentCacheValue = readOnlyCacheMap.get(key);
                       //从readWriteCacheMap取出对应Key的value，
                       //如果不一样以readWriteCacheMap的为准
                       if (cacheValue != currentCacheValue) {
                           readOnlyCacheMap.put(key, cacheValue);
                       }
                   } catch (Throwable th) {
                       logger.error("Error while updating the client cache from response cache", th);
                   }
               }
           }
       };
   }

由此可以看出，默认配置下从readOnlyCacheMap获取的实例信息，可能有30S的缓存时间数据是不一致的。

服务消费者缓存

服务消费者从eureka server获取实例信息并缓存在eureka client，如果用ribbon调用的话也会在ribbon内缓存一份实例信息。

Eureka Client缓存

Eureka Client的缓存放在com.netflix.discovery.shared.Applications#appNameApplicationMap中，appNameApplicationMap的数据结构是Map<String, Application>，key是appName。

应用启动初始化com.netflix.discovery.DiscoveryClient的时候会全量拉取一次服务信息，后面增量获取。

com.netflix.discovery.DiscoveryClient#initScheduledTasks

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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);
      }

registryFetchIntervalSeconds是eureka client从eureka server同步实例信息的时间，默认30S。

核心代码：com.netflix.discovery.DiscoveryClient#fetchRegistry

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
private boolean fetchRegistry(boolean forceFullRegistryFetch) {
    Stopwatch tracer = FETCH_REGISTRY_TIMER.start();

    try {
        // If the delta is disabled or if it is the first time, get all
        // applications
        Applications applications = getApplications();

        if (clientConfig.shouldDisableDelta()
                || (!Strings.isNullOrEmpty(clientConfig.getRegistryRefreshSingleVipAddress()))
                || forceFullRegistryFetch
                || (applications == null)
                || (applications.getRegisteredApplications().size() == 0)
                || (applications.getVersion() == -1)) //Client application does not have latest library supporting delta
        {
            logger.info("Disable delta property : {}", clientConfig.shouldDisableDelta());
            logger.info("Single vip registry refresh property : {}", clientConfig.getRegistryRefreshSingleVipAddress());
            logger.info("Force full registry fetch : {}", forceFullRegistryFetch);
            logger.info("Application is null : {}", (applications == null));
            logger.info("Registered Applications size is zero : {}",
                    (applications.getRegisteredApplications().size() == 0));
            logger.info("Application version is -1: {}", (applications.getVersion() == -1));
            //全量获取
            getAndStoreFullRegistry();
        } else {
            //增量获取
            getAndUpdateDelta(applications);
        }
        applications.setAppsHashCode(applications.getReconcileHashCode());
        logTotalInstances();
    } catch (Throwable e) {
        logger.error(PREFIX + appPathIdentifier + " - was unable to refresh its cache! status = " + e.getMessage(), e);
        return false;
    } finally {
        if (tracer != null) {
            tracer.stop();
        }
    }

    // Notify about cache refresh before updating the instance remote status
    onCacheRefreshed();

    // Update remote status based on refreshed data held in the cache
    updateInstanceRemoteStatus();

    // registry was fetched successfully, so return true
    return true;
}

Ribbon缓存

Ribbon的缓存放在com.netflix.loadbalancer.BaseLoadBalancer#allServerList中，应用启动的时候调用com.netflix.loadbalancer.PollingServerListUpdater#start启动一个更新ribbon缓存的定时任务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public synchronized void start(final UpdateAction updateAction) {
        if (isActive.compareAndSet(false, true)) {
            final Runnable wrapperRunnable = new Runnable() {
                @Override
                public void run() {
                    if (!isActive.get()) {
                        if (scheduledFuture != null) {
                            scheduledFuture.cancel(true);
                        }
                        return;
                    }
                    try {
                        updateAction.doUpdate();
                        lastUpdated = System.currentTimeMillis();
                    } catch (Exception e) {
                        logger.warn("Failed one update cycle", e);
                    }
                }
            };

            scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(
                    wrapperRunnable,
                    initialDelayMs,
                    refreshIntervalMs,
                    TimeUnit.MILLISECONDS
            );
        } else {
            logger.info("Already active, no-op");
        }
    }

refreshIntervalMs是ribbon缓存的刷新时间，可以配置单个应用的刷新时间aim-ms.ribbon.ServerListRefreshInterval，也可以配置全局的刷新时间ribbon.ServerListRefreshInterval，都不配置的话默认是30S。

获取配置代码:com.netflix.loadbalancer.PollingServerListUpdater

1
2
3
4
5
6
public PollingServerListUpdater(IClientConfig clientConfig) {
       this(LISTOFSERVERS_CACHE_UPDATE_DELAY, getRefreshIntervalMs(clientConfig));
   }
   private static long getRefreshIntervalMs(IClientConfig clientConfig) {
       return clientConfig.get(CommonClientConfigKey.ServerListRefreshInterval, LISTOFSERVERS_CACHE_REPEAT_INTERVAL);
   }

com.netflix.client.config.DefaultClientConfigImpl

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
public <T> T get(IClientConfigKey<T> key, T defaultValue) {
      T value = get(key);
      if (value == null) {
          value = defaultValue;
      }
      return value;
  }

  public <T> T get(IClientConfigKey<T> key) {
      Object obj = getProperty(key.key());
      if (obj == null) {
          return null;
      }
      Class<T> type = key.type();
      if (type.isInstance(obj)) {
          return type.cast(obj);
      } else {
          if (obj instanceof String) {
              String stringValue = (String) obj;
              if (Integer.class.equals(type)) {
                  return (T) Integer.valueOf(stringValue);
              } else if (Boolean.class.equals(type)) {
                  return (T) Boolean.valueOf(stringValue);
              } else if (Float.class.equals(type)) {
                  return (T) Float.valueOf(stringValue);
              } else if (Long.class.equals(type)) {
                  return (T) Long.valueOf(stringValue);
              } else if (Double.class.equals(type)) {
                  return (T) Double.valueOf(stringValue);
              } else if (TimeUnit.class.equals(type)) {
                  return (T) TimeUnit.valueOf(stringValue);
              }
              throw new IllegalArgumentException("Unable to convert string value to desired type " + type);
          }
           
          throw new IllegalArgumentException("Unable to convert value to desired type " + type);
      }
  }
  protected Object getProperty(String key) {
      if (enableDynamicProperties) {
          String dynamicValue = null;
          DynamicStringProperty dynamicProperty = dynamicProperties.get(key);
          if (dynamicProperty != null) {
              dynamicValue = dynamicProperty.get();
          }
          if (dynamicValue == null) {
              dynamicValue = DynamicProperty.getInstance(getConfigKey(key)).getString();
              if (dynamicValue == null) {
                  dynamicValue = DynamicProperty.getInstance(getDefaultPropName(key)).getString();
              }
          }
          if (dynamicValue != null) {
              return dynamicValue;
          }
      }
      return properties.get(key);
  }

刷新缓存代码：com.netflix.loadbalancer.BaseLoadBalancer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
public void setServersList(List lsrv) {
    Lock writeLock = allServerLock.writeLock();
    logger.debug("LoadBalancer [{}]: clearing server list (SET op)", name);
    
    ArrayList<Server> newServers = new ArrayList<Server>();
    writeLock.lock();
    try {
        ArrayList<Server> allServers = new ArrayList<Server>();
        for (Object server : lsrv) {
            if (server == null) {
                continue;
            }

            if (server instanceof String) {
                server = new Server((String) server);
            }

            if (server instanceof Server) {
                logger.debug("LoadBalancer [{}]:  addServer [{}]", name, ((Server) server).getId());
                allServers.add((Server) server);
            } else {
                throw new IllegalArgumentException(
                        "Type String or Server expected, instead found:"
                                + server.getClass());
            }

        }
        boolean listChanged = false;
        if (!allServerList.equals(allServers)) {
            listChanged = true;
            if (changeListeners != null && changeListeners.size() > 0) {
               List<Server> oldList = ImmutableList.copyOf(allServerList);
               List<Server> newList = ImmutableList.copyOf(allServers);                   
               for (ServerListChangeListener l: changeListeners) {
                   try {
                       l.serverListChanged(oldList, newList);
                   } catch (Exception e) {
                       logger.error("LoadBalancer [{}]: Error invoking server list change listener", name, e);
                   }
               }
            }
        }
        if (isEnablePrimingConnections()) {
            for (Server server : allServers) {
                if (!allServerList.contains(server)) {
                    server.setReadyToServe(false);
                    newServers.add((Server) server);
                }
            }
            if (primeConnections != null) {
                primeConnections.primeConnectionsAsync(newServers, this);
            }
        }
        // This will reset readyToServe flag to true on all servers
        // regardless whether
        // previous priming connections are success or not
        allServerList = allServers;
        if (canSkipPing()) {
            for (Server s : allServerList) {
                s.setAlive(true);
            }
            upServerList = allServerList;
        } else if (listChanged) {
            forceQuickPing();
        }
    } finally {
        writeLock.unlock();
    }
}

获取应用实例

eureka server入口：com.netflix.eureka.resources.ApplicationsResource#getContainers

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
@GET
public Response getContainers(@PathParam("version") String version,
                              @HeaderParam(HEADER_ACCEPT) String acceptHeader,
                              @HeaderParam(HEADER_ACCEPT_ENCODING) String acceptEncoding,
                              @HeaderParam(EurekaAccept.HTTP_X_EUREKA_ACCEPT) String eurekaAccept,
                              @Context UriInfo uriInfo,
                              @Nullable @QueryParam("regions") String regionsStr) {

    boolean isRemoteRegionRequested = null != regionsStr && !regionsStr.isEmpty();
    String[] regions = null;
    if (!isRemoteRegionRequested) {
        EurekaMonitors.GET_ALL.increment();
    } else {
        regions = regionsStr.toLowerCase().split(",");
        Arrays.sort(regions); // So we don't have different caches for same regions queried in different order.
        EurekaMonitors.GET_ALL_WITH_REMOTE_REGIONS.increment();
    }

    // Check if the server allows the access to the registry. The server can
    // restrict access if it is not
    // ready to serve traffic depending on various reasons.
    if (!registry.shouldAllowAccess(isRemoteRegionRequested)) {
        return Response.status(Status.FORBIDDEN).build();
    }
    CurrentRequestVersion.set(Version.toEnum(version));
    KeyType keyType = Key.KeyType.JSON;
    String returnMediaType = MediaType.APPLICATION_JSON;
    if (acceptHeader == null || !acceptHeader.contains(HEADER_JSON_VALUE)) {
        keyType = Key.KeyType.XML;
        returnMediaType = MediaType.APPLICATION_XML;
    }

    Key cacheKey = new Key(Key.EntityType.Application,
            ResponseCacheImpl.ALL_APPS,
            keyType, CurrentRequestVersion.get(), EurekaAccept.fromString(eurekaAccept), regions
    );

    Response response;
    if (acceptEncoding != null && acceptEncoding.contains(HEADER_GZIP_VALUE)) {
        response = Response.ok(responseCache.getGZIP(cacheKey))
                .header(HEADER_CONTENT_ENCODING, HEADER_GZIP_VALUE)
                .header(HEADER_CONTENT_TYPE, returnMediaType)
                .build();
    } else {
        response = Response.ok(responseCache.get(cacheKey))
                .build();
    }
    return response;
}

核心代码：com.netflix.eureka.registry.ResponseCacheImpl#getValue

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
Value getValue(final Key key, boolean useReadOnlyCache) {
        Value payload = null;
        try {
            if (useReadOnlyCache) {
                final Value currentPayload = readOnlyCacheMap.get(key);
                if (currentPayload != null) {
                    payload = currentPayload;
                } else {
                    payload = readWriteCacheMap.get(key);
                    readOnlyCacheMap.put(key, payload);
                }
            } else {
                payload = readWriteCacheMap.get(key);
            }
        } catch (Throwable t) {
            logger.error("Cannot get value for key :" + key, t);
        }
        return payload;
    }

如果useReadOnlyCache为true从readOnlyCacheMap中获取实例信息，如果false从readWriteCacheMap中获取实例信息，默认为true。

因为readWriteCacheMap用的LoadingCache有读写锁，使用readOnlyCacheMap可以增加吞吐量，中小集群可以关闭readOnlyCacheMap。

参考文档

https://blog.csdn.net/u012394095/article/details/80693713

http://blog.didispace.com/spring-cloud-eureka-register-detail/

https://www.infoq.cn/article/jlDJQ*3wtN2PcqTDyokh

java