SpringCloud 源码系列（2）—— 注册中心 Eureka（中）

 refreshInstanceInfo() {
 如果有必要，就更新数据中心的信息
 3     applicationInfoManager.refreshDataCenterInfoIfRequired();
 如果有必要，就更新续约信息，比如动态更新了配置文件，这时就更新续约信息 LeaseInfo，并将实例设置为 dirty
    applicationInfoManager.refreshLeaseInfoIfRequired();
 7     InstanceStatus status;
 8      9          用监控检查器检查实例的状态
10         status = getHealthCheckHandler().getStatus(instanceInfo.getStatus());
11     }  (Exception e) {
12         logger.warn("Exception from healthcheckHandler.getStatus,setting status to DOWN"13         status = InstanceStatus.DOWN;
14 15 
if (null != status) {
 设置实例状态，实例状态变了会触发状态变更的监听器
18         applicationInfoManager.setInstanceStatus(status);
21 
22 ////////////////////////////////23 
24  refreshLeaseInfoIfRequired() {
25      当前实例续约信息
26     LeaseInfo leaseInfo = instanceInfo.getLeaseInfo();
27     if (leaseInfo == 28         return30      从配置中获取续约信息
31     int currentLeaseDuration = config.getLeaseExpirationDurationInSeconds();
32     int currentLeaseRenewal = config.getLeaseRenewalIntervalInSeconds();
33      如果续约信息变了，就重新创建续约信息，并设置实例为 dirty
34     if (leaseInfo.getDurationInSecs() != currentLeaseDuration || leaseInfo.getRenewalIntervalInSecs() != currentLeaseRenewal) {
35         LeaseInfo newLeaseInfo = LeaseInfo.Builder.newBuilder()
36                 .setRenewalIntervalInSecs(currentLeaseRenewal)
37                 .setDurationInSecs(currentLeaseDuration)
                .build();
39         instanceInfo.setLeaseInfo(newLeaseInfo);
40 42 43 
44 45 
46  HealthCheckHandler getHealthCheckHandler() {
47     HealthCheckHandler healthCheckHandler = .healthCheckHandlerRef.get();
48     if (healthCheckHandler == 49          可以自定义 HealthCheckHandler 实现健康检查
50          healthCheckHandlerProvider) {
51             healthCheckHandler = healthCheckHandlerProvider.get();
52         } else  healthCheckCallbackProvider) {
53              可以自定义 HealthCheckCallback 实现健康检查，HealthCheckCallback 已过期，建议使用 HealthCheckHandler
54             healthCheckHandler =  HealthCheckCallbackToHandlerBridge(healthCheckCallbackProvider.get());
55 56 
57         null == healthCheckHandler) {
58              没有自定义的就是用默认的桥接类
59             healthCheckHandler = new HealthCheckCallbackToHandlerBridge(60 61         this.healthCheckHandlerRef.compareAndSet(62 63 
64     return 65 66 
67 //////////////////////////////////////
68 
69  setInstanceStatus(InstanceStatus status) {
70     InstanceStatus next = instanceStatusMapper.map(status);
71     if (next == 72         73 74 
75      如果状态变更了，才会返回之前的状态，然后触发状态变更监听器
76     InstanceStatus prev = instanceInfo.setStatus(next);
77     if (prev != 78         for (StatusChangeListener listener : listeners.values()) {
79             80                 listener.notify( StatusChangeEvent(prev,next));
81             } 82                 logger.warn("failed to notify listener: {}"83 84 85 86 }

boolean register() throws Throwable {
 2     logger.info(PREFIX + "{}: registering service..." 3     EurekaHttpResponse<Void> httpResponse;
 registrationClient => JerseyReplicationClient
 6         httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
 7     }  8         logger.warn(PREFIX + "{} - registration failed {}"throw e;
 (logger.isInfoEnabled()) {
12         logger.info(PREFIX + "{} - registration status: {}"return httpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode();
16 
17 18 
public EurekaHttpResponse<Void> register(InstanceInfo info) {
 调用的是 POST apps/{appName} 接口
21     String urlPath = "apps/" + info.getAppName();
22     ClientResponse response = 23     24         Builder resourceBuilder = jerseyClient.resource(serviceUrl).path(urlPath).getRequestBuilder();
25         addExtraHeaders(resourceBuilder);
26         response = resourceBuilder
27                 .header("Accept-Encoding","gzip"                .type(MediaType.APPLICATION_JSON_TYPE)
                .accept(MediaType.APPLICATION_JSON)
30                  post 方法
31                 .post(ClientResponse.class anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
33     } 34          (logger.isDebugEnabled()) {
35             logger.debug("Jersey HTTP POST {}/{} with instance {}; statusCode={}"36                     response == null ? "N/A" : response.getStatus());
38         if (response !=             response.close();
42 }

boolean onDemandUpdate() {
 限流控制
 (rateLimiter.acquire(burstSize,allowedRatePerMinute)) {
if (!scheduler.isShutdown()) {
 5             scheduler.submit( Runnable() {
 6                 @Override
 7                  8                     logger.debug("Executing on-demand update of local InstanceInfo" 9 
10                      如果上一次的任务还没有执行完，直接取消掉，然后执行注册的任务
11                     Future latestPeriodic = scheduledPeriodicRef.get();
12                     if (latestPeriodic != null && !latestPeriodic.isDone()) {
13                         logger.debug("Canceling the latest scheduled update,it will be rescheduled at the end of on demand update"14                         latestPeriodic.cancel(                    }
17                     InstanceInfoReplicator..run();
            });
20             21         } 22             logger.warn("Ignoring onDemand update due to stopped scheduler"23             25     } 26         logger.warn("Ignoring onDemand update due to rate limiter"27         29 }

/**
 2  * Rate limiter implementation is based on token bucket algorithm. There are two parameters:
 * <ul>
 * <li>
 *     burst size - maximum number of requests allowed into the system as a burst
 * </li>
 8  *     average rate - expected number of requests per second (RateLimiters using MINUTES is also supported)
 * </ul>
 *
 * @author Tomasz Bak
13  */
 RateLimiter {
final long rateToMsConversion;
17 
final AtomicInteger consumedTokens =  AtomicInteger();
final AtomicLong lastRefillTime = new AtomicLong(0    @Deprecated
22      RateLimiter() {
23         (TimeUnit.SECONDS);
25 
26      RateLimiter(TimeUnit averageRateUnit) {
switch (averageRateUnit) {
28             case SECONDS:
29                 rateToMsConversion = 1000break31              MINUTES:
32                 rateToMsConversion = 60 * 100033                 34             default:
35                 throw new IllegalArgumentException("TimeUnit of " + averageRateUnit + " is not supported"38 
39     boolean acquire(int burstSize,1)"> averageRate) {
40          acquire(burstSize,averageRate,System.currentTimeMillis());
42 
43     long averageRate,1)"> currentTimeMillis) {
44         if (burstSize <= 0 || averageRate <= 0) {  Instead of throwing exception,we just let all the traffic go
45             47 
48         refillToken(burstSize,currentTimeMillis);
 consumeToken(burstSize);
50 51 
52     void refillToken(53          上一次填充 token 的时间
54         long refillTime = lastRefillTime.get();
55          时间差
56         long timeDelta = currentTimeMillis - refillTime;
 固定生成令牌的速率，即每分钟4次
58          例如刚好间隔15秒进来一个请求，就是 15000 * 4 / 60000 = 1，newTokens 代表间隔了多少次，如果等于0，说明间隔不足15秒
59         long newTokens = timeDelta * averageRate /60         if (newTokens > 061             long newRefillTime = refillTime == 0
62                     ? currentTimeMillis
63                      注意这里不是直接设置的当前时间戳，而是根据 newTokens 重新计算的，因为有可能同一周期内同时有多个请求进来，这样可以保持一个固定的周期
64                     : refillTime + newTokens * rateToMsConversion / averageRate;
65              (lastRefillTime.compareAndSet(refillTime,newRefillTime)) {
66                 while (67                      调整令牌的数量
68                     int currentLevel = consumedTokens.get();
69                     int adjustedLevel = Math.min(currentLevel,burstSize);
70                      currentLevel 可能为2，重置为了 0 或 1
71                     int newLevel = (int) Math.max(0,adjustedLevel - newTokens);
72                      (consumedTokens.compareAndSet(currentLevel,newLevel)) {
73                         74 75 76 77 78 79 
80     boolean consumeToken(81         82             83              突发数量为2，也就是允许15秒内最多有两次请求进来
84             if (currentLevel >=85                 86 87             if (consumedTokens.compareAndSet(currentLevel,currentLevel + 188                 89 90 91 92 
93      reset() {
94         consumedTokens.set(095         lastRefillTime.set(096 97 }

void register(InstanceInfo registrant,1)">int leaseDuration,1)">    read.lock();
 registry => ConcurrentHashMap<String,Lease<InstanceInfo>>>
 5         Map<String,Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
        REGISTER.increment(isReplication);
if (gMap ==  8              初次注册时，创建一个 ConcurrentHashMap，key 为 appName
final ConcurrentHashMap<String,Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String,Lease<InstanceInfo>>10             gMap = registry.putIfAbsent(registrant.getAppName(),gNewMap);
12                 gMap = gNewMap;
15         Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
16          Retain the last dirty timestamp without overwriting it,if there is already a lease
if (existingLease != null && (existingLease.getHolder() != 18              已存在的实例的最后更新时间
19             Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
 新注册的实例的最后更新时间
21             Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
22             logger.debug("Existing lease found (existing={},provided={}"24              如果存在的实例比新注册尽量的实例后更新，就直接把新注册的实例设置为已存在的实例
25             if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
26                 logger.warn("There is an existing lease and the existing lease's dirty timestamp {} is greater" +
27                         " than the one that is being registered {}"28                 logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant"29                 registrant = existingLease.getHolder();
31         } 32              新注册时，续约信息不存在
33             synchronized (lock) {
34                 this.expectedNumberOfClientsSendingRenews > 035                      Since the client wants to register it,increase the number of clients sending renews
36                      期望续约的客户端数量 + 1
37                     this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 138                      更新每分钟续约请求次数的阀值，这个阀值在后面很多地方都会用到
                    updateRenewsPerMinThreshold();
42             logger.debug("No previous lease information found; it is new registration"43  创建新的续约
45         Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant,leaseDuration);
46         47             lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
49         gMap.put(registrant.getId(),lease);
 放入最近注册的队列
51         recentRegisteredQueue.add(new Pair<Long,String>(
52                 System.currentTimeMillis(),1)">53                 registrant.getAppName() + "(" + registrant.getId() + ")"));
 覆盖状态
InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) {
56             logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the "
57                             + "overrides"overriddenInstanceStatusMap.containsKey(registrant.getId())) {
59                 logger.info("Not found overridden id {} and hence adding it"                overriddenInstanceStatusMap.put(registrant.getId(),registrant.getOverriddenStatus());
61 63         InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());
64         if (overriddenStatusFromMap != 65             logger.info("Storing overridden status {} from map"66             registrant.setOverriddenStatus(overriddenStatusFromMap);
69          Set the status based on the overridden status rules
70         InstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant,existingLease,1)">71          仅仅是变更实例状态，不会设置为 dirty
72         registrant.setStatusWithoutDirty(overriddenInstanceStatus);
73 
74          If the lease is registered with UP status,set lease service up timestamp
75          (InstanceStatus.UP.equals(registrant.getStatus())) {
76              UP 时设置 Lease 的时间戳
            lease.serviceUp();
79          设置动作是 ADDED，这个在后面会做 switch 判断
80         registrant.setActionType(ActionType.ADDED);
 添加到最近变更的队列
82         recentlyChangedQueue.add( RecentlyChangedItem(lease));
83          设置最后更新时间
        registrant.setLastUpdatedTimestamp();
85          失效缓存
        invalidateCache(registrant.getAppName(),registrant.getVIPAddress(),registrant.getSecureVipAddress());
87         logger.info("Registered instance {}/{} with status {} (replication={})"88                 registrant.getAppName(),registrant.getId(),registrant.getStatus(),1)">89     }         read.unlock();
92 }

  1 <%@ page language="java" import="java.util.*,java.util.Map.Entry,com.netflix.discovery.shared.Pair,1)">  2 com.netflix.discovery.shared.*,com.netflix.eureka.util.*,com.netflix.appinfo.InstanceInfo.*  3 com.netflix.appinfo.DataCenterInfo.*,com.netflix.appinfo.AmazonInfo.MetaDataKey,com.netflix.eureka.resources.*  4 com.netflix.eureka.*,com.netflix.appinfo.*,com.netflix.eureka.util.StatusUtil" pageEncoding="UTF-8" %>
  5 <%
  6 String path = request.getContextPath();
  7 String basePath = request.getScheme()+"://"+request.getServerName()+":"+request.getServerPort()+path+"/"  8 %>
  9 
 10 <!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">
 11 
 12 <html>
 13   <head>
 14     <base href="<%=basePath%>">
 15 
 16     <title>Eureka</title>
 17     <link rel="stylesheet" type="text/css" href="./css/main.css">
 18     <script type="text/javascript" src="./js/jquery-1.11.1.js" ></script>
 19     <script type="text/javascript" src="./js/jquery.dataTables.js" ></script>
 20     <script type="text/javascript" >
 21        $(document).ready(function() {
 22            $('table.stripeable tr:odd').addClass('odd' 23            $('table.stripeable tr:even').addClass('even' 24            $('#instances thead th').each(function () {
 25                var title = $('#instances thead th').eq($().index()).text();
 26                $(this).html(title + '</br><input type="text" placeholder="Search ' + title + '" />' 27            });
 28             DataTable
 29            var table = $('#instances').DataTable({"paging": });
 30             Apply the search
 31            table.columns().eq(0).each(function (colIdx) {
 32                $('input',table.column(colIdx).header()).on('keyup change' 33                    table.column(colIdx).search(.value).draw();
 34                });
 35  36        });
 37     </script>
 38   </head>
 39 
 40   <body id="one">
 41     <jsp:include page="header.jsp" />
 42     <jsp:include page="navbar.jsp" />
 43     <div id="content">
 44       <div class="sectionTitle">Instances currently registered with Eureka</div>
 45         <table id='instances' class="stripeable">
 46            <thead><tr><th>Application</th><th>AMIs</th><th>Availability Zones</th><th>Status</th></tr></thead>
 47            <tfoot><tr><th>Application</th><th>AMIs</th><th>Availability Zones</th><th>Status</th></tr></tfoot>
 48            <tbody>
 49            <%
 50                 获取 eureka server 上下文 EurekaServerContext
 51            EurekaServerContext serverContext = (EurekaServerContext) pageContext.getServletContext()
 52                    .getAttribute(EurekaServerContext..getName());
 53             从上下文中取出注册表，
 54            (Application app : serverContext.getRegistry().getSortedApplications()) {
 55                out.print("<tr><td><b>" + app.getName() + "</b></td>" 56                Map<String,Integer> amiCounts = new HashMap<String,Integer> 57                Map<InstanceStatus,List<Pair<String,String>>> instancesByStatus =
 58                    new HashMap<InstanceStatus,String>>> 59                Map<String,Integer> zoneCounts =  60 
 61                (InstanceInfo info : app.getInstances()){
 62                    String id = info.getId();
 63                    String url = info.getStatusPageUrl();
 64                    InstanceStatus status = info.getStatus();
 65                    String ami = "n/a" 66                    String zone = "" 67                    if(info.getDataCenterInfo().getName() == Name.Amazon){
 68                        AmazonInfo dcInfo = (AmazonInfo)info.getDataCenterInfo();
 69                        ami = dcInfo.get(MetaDataKey.amiId);
 70                        zone = dcInfo.get(MetaDataKey.availabilityZone);
 71                    }
 72 
 73                    Integer count = amiCounts.get(ami);
 74                    if(count != ){
 75                        amiCounts.put(ami,Integer.valueOf(count.intValue()+1 76                    } 77                        amiCounts.put(ami,Integer.valueOf(1 78  79 
 80                    count = zoneCounts.get(zone);
 81                     82                        zoneCounts.put(zone,1)"> 83                    } 84                        zoneCounts.put(zone,1)"> 85  86                    List<Pair<String,String>> list = instancesByStatus.get(status);
 87 
 88                    if(list ==  89                        list = new ArrayList<Pair<String,String>> 90                        instancesByStatus.put(status,list);
 91  92                    list.add(new Pair<String,1)">(id,url));
 93                }
 94                StringBuilder buf =  StringBuilder();
 95                for (Iterator<Entry<String,Integer>> iter =
 96                    amiCounts.entrySet().iterator(); iter.hasNext();) {
 97                    Entry<String,Integer> entry = iter.next();
 98                    buf.append("<b>").append(entry.getKey()).append("</b> (").append(entry.getValue()).append(")," 99 100                out.println("<td>" + buf.toString() + "</td>"101                buf = 102                103                    zoneCounts.entrySet().iterator(); iter.hasNext();) {
104                    Entry<String,1)">105                    buf.append("<b>").append(entry.getKey()).append("</b> (").append(entry.getValue()).append("),1)">106 107                out.println("<td>" + buf.toString() + "</td>"108                buf = 109                for (Iterator<Entry<InstanceStatus,String>>>> iter =
110                    instancesByStatus.entrySet().iterator(); iter.hasNext();) {
111                    Entry<InstanceStatus,String>>> entry =112                    List<Pair<String,String>> value = entry.getValue();
113                    InstanceStatus status = entry.getKey();
114                    if(status != InstanceStatus.UP){
115                        buf.append("<font color=red size=+1><b>"116 117                    buf.append("<b>").append(status.name()).append("</b> (").append(value.size()).append(") - "118                    119                        buf.append("</font></b>"120 121 
122                    for(Pair<String,1)"> p : value) {
123                        String id = p.first();
124                        String url = p.second();
125                        if(url != null && url.startsWith("http")){
126                            buf.append("<a href="").append(url).append("">"127                        }128                            url = 129                        }
130                        buf.append(id);
131                        132                            buf.append("</a>"133 134                        buf.append(",1)">135 136 137                out.println("<td>" + buf.toString() + "</td></tr>"138            }
139            %>
140            </tbody>
141            </table>
142       </div>
143       <div>
144       <div class="sectionTitle">General Info</div>
145       <table id='generalInfo' 146           <tr><th>Name</th><th>Value</th></tr>
147            <%
148            StatusInfo statusInfo = ( StatusUtil(serverContext)).getStatusInfo();
149            Map<String,String> genMap = statusInfo.getGeneralStats();
150            for (Map.Entry<String,1)"> entry : genMap.entrySet()) {
151              out.print("<tr>"152              out.print("<td>" + entry.getKey() +  "</td><td>" + entry.getValue() + "</td>"153              out.print("</tr>"154 155            Map<String,String> appMap = statusInfo.getApplicationStats();
156             entry : appMap.entrySet()) {
157              out.print("<tr>"158              out.print("<td>" + entry.getKey() +  "</td><td>" + entry.getValue() + "</td>"159              out.print("</tr>"160 161            %>
162            </table>
163       </div>
164       <div>
165       <div class="sectionTitle">Instance Info</div>
166         <table id='instanceInfo' 167           <tr><th>Name</th><th>Value</th></tr>
168            <%
169            InstanceInfo instanceInfo = statusInfo.getInstanceInfo();
170            Map<String,String> instanceMap = 171            instanceMap.put("ipAddr"172            instanceMap.put("status"173            if(instanceInfo.getDataCenterInfo().getName() == DataCenterInfo.Name.Amazon) {
174                AmazonInfo info = (AmazonInfo) instanceInfo.getDataCenterInfo();
175                instanceMap.put("availability-zone"176                instanceMap.put("public-ipv4"177                instanceMap.put("instance-id"178                instanceMap.put("public-hostname"179                instanceMap.put("ami-id"180                instanceMap.put("instance-type"181 182             entry : instanceMap.entrySet()) {
183              out.print("<tr>"184              out.print("<td>" + entry.getKey() +  "</td><td>" + entry.getValue() + "</td>"185              out.print("</tr>"186 187            %>
188            </table>
189     </div>
190 
191   </body>
192 </html>

boolean fetchRegistry( forceFullRegistryFetch) {
 2     Stopwatch tracer = FETCH_REGISTRY_TIMER.start();
 获取本地的应用实例
 6         Applications applications = getApplications();
 (clientConfig.shouldDisableDelta()
 9                 || (!Strings.isNullOrEmpty(clientConfig.getRegistryRefreshSingleVipAddress()))
10                 || forceFullRegistryFetch
11                 || (applications == 12                 || (applications.getRegisteredApplications().size() == 013                 || (applications.getVersion() == -1)) Client application does not have latest library supporting delta
        {
 全量抓取注册表
            getAndStoreFullRegistry();
 增量更新注册表
            getAndUpdateDelta(applications);
        applications.setAppsHashCode(applications.getReconcileHashCode());
        logTotalInstances();
23     }  (Throwable e) {
24         logger.info(PREFIX + "{} - was unable to refresh its cache! This periodic background refresh will be retried in {} seconds. status = {} stacktrace = {}"                appPathIdentifier,clientConfig.getRegistryFetchIntervalSeconds(),ExceptionUtils.getStackTrace(e));
26         27     } if (tracer !=             tracer.stop();
31 32 
 发出缓存刷新的通知
    onCacheRefreshed();
36      Update remote status based on refreshed data held in the cache
    updateInstanceRemoteStatus();
 registry was fetched successfully,so return true
40     41 }

41 }

private Value generatePayload(Key key) {
 2     Stopwatch tracer =         String payload;
 (key.getEntityType()) {
 Application:
boolean isRemoteRegionRequested = key.hasRegions();
 8 
 获取所有应用
10                  (ALL_APPS.equals(key.getName())) {
11                      (isRemoteRegionRequested) {
12                         tracer = serializeAllAppsWithRemoteRegionTimer.start();
13                         payload = getPayLoad(key,registry.getApplicationsFromMultipleRegions(key.getRegions()));
14                     } 15                         tracer = serializeAllAppsTimer.start();
16                          从注册表读取所有服务实例
17                         payload =20                  增量获取应用
21                  (ALL_APPS_DELTA.equals(key.getName())) {
22                     23                         tracer = serializeDeltaAppsWithRemoteRegionTimer.start();
                        versionDeltaWithRegions.incrementAndGet();
                        versionDeltaWithRegionsLegacy.incrementAndGet();
26                         payload =                                registry.getApplicationDeltasFromMultipleRegions(key.getRegions()));
28                     } 29                         tracer = serializeDeltaAppsTimer.start();
                        versionDelta.incrementAndGet();
                        versionDeltaLegacy.incrementAndGet();
32                         payload =34                 } 35                     tracer = serializeOneApptimer.start();
36                     payload =38                 39              VIP:
40              SVIP:
41                 tracer = serializeViptimer.start();
42                 payload =43                 44             45                 logger.error("Unidentified entity type: {} found in the cache key."46                 payload = ""47                  Value(payload);
50     } 51         54 55 }

 invalidate(String appName,1)"> (Key.KeyType type : Key.KeyType.values()) {
 (Version v : Version.values()) {
            invalidate(
 6                      失效服务的缓存
 7                      8                      9                      失效所有 APP 的缓存
 失效增量 APP 的缓存
13                                 );
 vipAddress) {
17                 invalidate( Key(Key.EntityType.VIP,EurekaAccept.full));
19              secureVipAddress) {
20                 invalidate( Key(Key.EntityType.SVIP,secureVipAddress,1)"> invalidate(Key... keys) {
 (Key key : keys) {
28         logger.debug("Invalidating the response cache key : {} {} {} {},{}"                key.getEntityType(),key.getName(),key.getVersion(),key.getType(),key.getEurekaAccept());
30 
31          失效读写缓存
        readWriteCacheMap.invalidate(key);
33         Collection<Key> keysWithRegions = regionSpecificKeys.get(key);
null != keysWithRegions && !keysWithRegions.isEmpty()) {
 (Key keysWithRegion : keysWithRegions) {
36                 logger.debug("Invalidating the response cache key : {} {} {} {} {}"                        key.getEntityType(),1)">                readWriteCacheMap.invalidate(keysWithRegion);
42 }

 TimerTask getCacheUpdateTask() {
 TimerTask() {
        @Override
 5             logger.debug("Updating the client cache from response cache" (Key key : readOnlyCacheMap.keySet()) {
 8                     logger.debug("Updating the client cache from response cache for key : {} {} {} {}"                            key.getEntityType(),key.getType());
11                                     CurrentRequestVersion.set(key.getVersion());
 获取读写缓存中的数据
14                     Value cacheValue =15                      获取只读缓存中的数据
16                     Value currentCacheValue =17                      如果 readOnlyCacheMap 中缓存的值与 readWriteCacheMap 缓存的值不同，就用 readWriteCacheMap 的值覆盖 readOnlyCacheMap 的值
18                     if (cacheValue != currentCacheValue) {
                        readOnlyCacheMap.put(key,cacheValue);
21                 } 22                     logger.error("Error while updating the client cache from response cache for key {}"23                 }                     CurrentRequestVersion.remove();
    };
29 }

class CacheRefreshThread implements Runnable {
        refreshRegistry();
@VisibleForTesting
 refreshRegistry() {
10         boolean isFetchingRemoteRegionRegistries = isFetchingRemoteRegionRegistries();
11 
12         boolean remoteRegionsModified =  This makes sure that a dynamic change to remote regions to fetch is honored.
14         String latestRemoteRegions = clientConfig.fetchRegistryForRemoteRegions();
15          latestRemoteRegions) {
16             String currentRemoteRegions = remoteRegionsToFetch.get();
latestRemoteRegions.equals(currentRemoteRegions)) {
 Both remoteRegionsToFetch and AzToRegionMapper.regionsToFetch need to be in sync
19                  (instanceRegionChecker.getAzToRegionMapper()) {
20                      (remoteRegionsToFetch.compareAndSet(currentRemoteRegions,latestRemoteRegions)) {
21                         String[] remoteRegions = latestRemoteRegions.split(",1)">                        remoteRegionsRef.set(remoteRegions);
                        instanceRegionChecker.getAzToRegionMapper().setRegionsToFetch(remoteRegions);
24                         remoteRegionsModified = 25                     } 26                         logger.info("Remote regions to fetch modified concurrently," +
27                                 " ignoring change from {} to {}"30             } 31                  Just refresh mapping to reflect any DNS/Property change
                instanceRegionChecker.getAzToRegionMapper().refreshMapping();
36          抓取注册表
boolean success = fetchRegistry(remoteRegionsModified);
 (success) {
39             registrySize = localRegionApps.get().size();
40             lastSuccessfulRegistryFetchTimestamp =43         44             StringBuilder allAppsHashCodes = 45             allAppsHashCodes.append("Local region apps hashcode: "            allAppsHashCodes.append(localRegionApps.get().getAppsHashCode());
47             allAppsHashCodes.append(",is fetching remote regions? "            allAppsHashCodes.append(isFetchingRemoteRegionRegistries);
49              entry : remoteRegionVsApps.entrySet()) {
50                 allAppsHashCodes.append(",Remote region: "51                 allAppsHashCodes.append(entry.getKey());
52                 allAppsHashCodes.append(",apps hashcode: "                allAppsHashCodes.append(entry.getValue().getAppsHashCode());
55             logger.debug("Completed cache refresh task for discovery. All Apps hash code is {} "56                     allAppsHashCodes);
57 58     } 59         logger.error("Cannot fetch registry from server"61 }

15             logger.info("Disable delta property : {}"16             logger.info("Single vip registry refresh property : {}"17             logger.info("Force full registry fetch : {}"18             logger.info("Application is null : {}",(applications == 19             logger.info("Registered Applications size is zero : {}"20                     (applications.getRegisteredApplications().size() == 021             logger.info("Application version is -1: {}",(applications.getVersion() == -124         } 30     } 31         logger.info(PREFIX + "{} - was unable to refresh its cache! This periodic background refresh will be retried in {} seconds. status = {} stacktrace = {}"34     } 35          发出刷新缓存的通知
46     47     48 }

 updateDelta(Applications delta) {
int deltaCount = 0 变量增量注册的服务
 (Application app : delta.getRegisteredApplications()) {
 遍历实例
 (InstanceInfo instance : app.getInstances()) {
 7             Applications applications = 8             String instanceRegion = instanceRegionChecker.getInstanceRegion(instance);
instanceRegionChecker.isLocalRegion(instanceRegion)) {
10                 Applications remoteApps = remoteRegionVsApps.get(instanceRegion);
 remoteApps) {
12                     remoteApps =  Applications();
                    remoteRegionVsApps.put(instanceRegion,remoteApps);
15                 applications = remoteApps;
18             ++deltaCount;
 ADDED 新增的实例：服务注册
 (ActionType.ADDED.equals(instance.getActionType())) {
21                 Application existingApp = applications.getRegisteredApplications(instance.getAppName());
22                 if (existingApp ==                     applications.addApplication(app);
25                 logger.debug("Added instance {} to the existing apps in region {}"                applications.getRegisteredApplications(instance.getAppName()).addInstance(instance);
 MODIFIED 变更的实例：续约，信息变更
29              (ActionType.MODIFIED.equals(instance.getActionType())) {
30                 Application existingApp =34                 logger.debug("Modified instance {} to the existing apps "37 
 DELETED 移除实例：实例下线、故障
 (ActionType.DELETED.equals(instance.getActionType())) {
41                 Application existingApp =42                 if (existingApp != 43                     logger.debug("Deleted instance {} to the existing apps "                    existingApp.removeInstance(instance);
45                     /*
                     * We find all instance list from application(The status of instance status is not only the status is UP but also other status)
                     * if instance list is empty,we remove the application.
48                      49                      (existingApp.getInstancesAsIsFromEureka().isEmpty()) {
                        applications.removeApplication(existingApp);
56     logger.debug("The total number of instances fetched by the delta processor : {}"57 
58     getApplications().setVersion(delta.getVersion());
59     getApplications().shuffleInstances(clientConfig.shouldFilterOnlyUpInstances());
60 
61      (Applications applications : remoteRegionVsApps.values()) {
        applications.setVersion(delta.getVersion());
63         applications.shuffleInstances(clientConfig.shouldFilterOnlyUpInstances());
64 65 }

32                          获取增量注册表
33                         payload =35                 } 36                     tracer =37                     payload =39                 41             42                 tracer =43                 payload =46                 logger.error("Unidentified entity type: {} found in the cache key."47                 payload = ""48                 52         56 }

@PUT
 Response renewLease(
        @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication,1)"> 4         @QueryParam("overriddenstatus") String overriddenStatus,1)"> 5         @QueryParam("status") String status,1)"> 6         @QueryParam("lastDirtyTimestamp") String lastDirtyTimestamp) {
boolean isFromReplicaNode = "true".equals(isReplication);
 调用注册表的 renew 进行服务续约
boolean isSuccess = registry.renew(app.getName(),id,isFromReplicaNode);
 Not found in the registry,immediately ask for a register
isSuccess) {
13         logger.warn("Not Found (Renew): {} - {}" Response.status(Status.NOT_FOUND).build();
 Check if we need to sync based on dirty time stamp,the client
 instance might have changed some value
if (lastDirtyTimestamp !=  serverConfig.shouldSyncWhenTimestampDiffers()) {
20         response = .validateDirtyTimestamp(Long.valueOf(lastDirtyTimestamp),1)"> Store the overridden status since the validation found out the node that replicates wins
if (response.getStatus() == Response.Status.NOT_FOUND.getStatusCode()
23                 && (overriddenStatus != 24                 && !(InstanceStatus.UNKNOWN.name().equals(overriddenStatus))
25                 && isFromReplicaNode) {
            registry.storeOverriddenStatusIfRequired(app.getAppName(),InstanceStatus.valueOf(overriddenStatus));
28     } 29         response = Response.ok().build();
31     logger.debug("Found (Renew): {} - {}; reply status={}"33 }

void openForTraffic(ApplicationInfoManager applicationInfoManager,1)"> count) {
 期望的客户端每分钟的续约次数
this.expectedNumberOfClientsSendingRenews = count;
 更新每分钟续约阈值
    updateRenewsPerMinThreshold();
 6     logger.info("Got {} instances from neighboring DS node" 7     logger.info("Renew threshold is: {}"this.startupTime =if (count > 0this.peerInstancesTransferEmptyOnStartup = 12     DataCenterInfo.Name selfName = applicationInfoManager.getInfo().getDataCenterInfo().getName();
boolean isAws = Name.Amazon == selfName;
if (isAws && serverConfig.shouldPrimeAwsReplicaConnections()) {
15         logger.info("Priming AWS connections for all replicas.."        primeAwsReplicas(applicationInfoManager);
18     logger.info("Changing status to UP" 设置实例状态为已启动
    applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
 调用父类的后置初始化
.postInit();
23 }