SpringCloud 源碼系列（2）—— 注冊中心 Eureka（中）

SpringCloud 源碼系列（1）—— 注冊中心 Eureka（上）

SpringCloud 源碼系列（2）—— 注冊中心 Eureka（中）

SpringCloud 源碼系列（3）—— 注冊中心 Eureka（下）

 

五、服務注冊

1、實例信息注冊器初始化

服務注冊的代碼位置不容易發現，我們看 DiscoveryClient 初始化調度任務的這個方法，這段代碼會去初始化一個實例信息復制器 InstanceInfoReplicator，這個復制器就包含了實例的注冊（明明是注冊卻叫 Replicator 感覺怪怪的）。

① DiscoveryClient 初始化調度器的流程

• 先基於 DiscoveryClient、InstanceInfo 構造 InstanceInfoReplicator，然后還有兩個參數為實例信息復制間隔時間（默認30秒）、並發的數量（默認為2）。
• 創建了一個實例狀態變更監聽器，並注冊到 ApplicationInfoManager。當實例狀態變更時，就會觸發這個監聽器，並調用 InstanceInfoReplicator 的 onDemandUpdate 方法。
• 啟動 InstanceInfoReplicator，默認延遲40秒，也就是說服務啟動可能40秒之后才會注冊到注冊中心。

private void initScheduledTasks() {
    // 省略定時刷新注冊表的任務...

    if (clientConfig.shouldRegisterWithEureka()) {
        // 省略定時心跳的任務...

        // 實例信息復制器，用於定時更新自己狀態，並向注冊中心注冊
        instanceInfoReplicator = new InstanceInfoReplicator(
                this,
                instanceInfo,
                clientConfig.getInstanceInfoReplicationIntervalSeconds(),
                2); // burstSize

        // 實例狀態變更的監聽器
        statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
            @Override
            public String getId() {
                return "statusChangeListener";
            }

            @Override
            public void notify(StatusChangeEvent statusChangeEvent) {
                if (statusChangeEvent.getStatus() == InstanceStatus.DOWN) {
                    logger.error("Saw local status change event {}", statusChangeEvent);
                } else {
                    logger.info("Saw local status change event {}", statusChangeEvent);
                }
                instanceInfoReplicator.onDemandUpdate();
            }
        };

        // 向 ApplicationInfoManager 注冊狀態變更監聽器
        if (clientConfig.shouldOnDemandUpdateStatusChange()) {
            applicationInfoManager.registerStatusChangeListener(statusChangeListener);
        }

        // 啟動實例信息復制器，默認延遲時間40秒
        instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
    } else {
        logger.info("Not registering with Eureka server per configuration");
    }
}

② InstanceInfoReplicator 的構造方法

• 創建了一個單線程的調度器
• 設置 started 為 false
• 創建了以分鍾為單位的限流器，每分鍾默認最多只能調度4次

InstanceInfoReplicator(DiscoveryClient discoveryClient, InstanceInfo instanceInfo, int replicationIntervalSeconds, int burstSize) {
    this.discoveryClient = discoveryClient;
    this.instanceInfo = instanceInfo;
    // 單線程的調度器
    this.scheduler = Executors.newScheduledThreadPool(1,
            new ThreadFactoryBuilder()
                    .setNameFormat("DiscoveryClient-InstanceInfoReplicator-%d")
                    .setDaemon(true)
                    .build());

    this.scheduledPeriodicRef = new AtomicReference<Future>();
    // started 設置為 false
    this.started = new AtomicBoolean(false);
    // 以分鍾為單位的限流器
    this.rateLimiter = new RateLimiter(TimeUnit.MINUTES);
    // 間隔時間，默認為30秒
    this.replicationIntervalSeconds = replicationIntervalSeconds;
    this.burstSize = burstSize;
    // 允許每分鍾更新的頻率 60 * 2 / 30 = 4
    this.allowedRatePerMinute = 60 * this.burstSize / this.replicationIntervalSeconds;
    logger.info("InstanceInfoReplicator onDemand update allowed rate per min is {}", allowedRatePerMinute);
}

③ 啟動 InstanceInfoReplicator

• 將 started 設置為 true，代表已經啟動了
• 調用 instanceInfo.setIsDirty() 方法，將實例設置為 dirty=true，並更新了最后一次設置 dirty 的時間戳
• InstanceInfoReplicator 實現了 Runnable，它本身被當成任務來調度，然后延遲40秒開始調度當前任務，並將 Future 放到本地變量中

public void start(int initialDelayMs) {
    // 啟動時 started 設置為 true
    if (started.compareAndSet(false, true)) {
        // 設置為 dirty，便於下一次心跳時同步到 eureka server
        instanceInfo.setIsDirty();
        // 延遲40秒后開始調度當前任務
        Future next = scheduler.schedule(this, initialDelayMs, TimeUnit.SECONDS);
        // 將 Future 放到本地變量中
        scheduledPeriodicRef.set(next);
    }
}

///////

public synchronized void setIsDirty() {
    isInstanceInfoDirty = true;
    lastDirtyTimestamp = System.currentTimeMillis();
}

2、客戶端實例注冊

① 實現注冊的run方法

接着看 InstanceInfoReplicator 的 run 方法，這個方法就是完成注冊的核心位置。

• 首先會更新實例的信息，如果有變更就會設置 dirty=true
• 如過是 dirty 的，就會調用 DiscoveryClient 的 register 方法注冊實例
• 實例注冊后，就把 dirty 設置為 false
• 最后在 finally 中繼續下一次的調度，默認是每隔30秒調度一次，注意他這里是把調度結果 Future 放到本地變量中

public void run() {
    try {
        // 更新本地實例信息，如果實例信息有變更，則 dirty=true
        discoveryClient.refreshInstanceInfo();

        // 設置為 dirty 時的時間戳
        Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
        if (dirtyTimestamp != null) {
            // 注冊實例
            discoveryClient.register();
            // 設置 dirty=false
            instanceInfo.unsetIsDirty(dirtyTimestamp);
        }
    } catch (Throwable t) {
        logger.warn("There was a problem with the instance info replicator", t);
    } finally {
        // 30秒之后再調度
        Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
        scheduledPeriodicRef.set(next);
    }
}

② 實例信息刷新

再來細看下 refreshInstanceInfo 刷新實例信息的方法：

• 首先刷新了數據中心的信息
• 然后刷新續約信息，主要就是將 EurekaClientConfig 的續約配置與本地的續約配置做對比，如果變更了就重新創建續約信息，並設置實例為dirty。這種情況一般就是運行期間動態更新實例的配置，然后重新注冊實例信息。
• 接着使用健康檢查器檢查實例健康狀況，從 getHealthCheckHandler 這段代碼進去不難發現，我們可以自定義健康檢查器，例如當本地的一些資源未創建成功、某些核心線程池down了就認為實例不可用，這個時候就可以自定義健康檢查器。如果沒有自定義健康檢查器，那就直接返回實例當前的狀態。我們可以實現 HealthCheckHandler 接口自定義健康檢查器。
• 最后就會調用 ApplicationInfoManager 的 setInstanceStatus 設置實例狀態，會判斷如果狀態發生變更，就會發出狀態變更的通知，這樣就會觸發前面定義的狀態變更監聽器，然后調用 InstanceInfoReplicator 的 onDemandUpdate 方法。

void refreshInstanceInfo() {
    // 如果有必要，就更新數據中心的信息
    applicationInfoManager.refreshDataCenterInfoIfRequired();
    // 如果有必要，就更新續約信息，比如動態更新了配置文件，這時就更新續約信息 LeaseInfo，並將實例設置為 dirty
    applicationInfoManager.refreshLeaseInfoIfRequired();

    InstanceStatus status;
    try {
        // 用監控檢查器檢查實例的狀態
        status = getHealthCheckHandler().getStatus(instanceInfo.getStatus());
    } catch (Exception e) {
        logger.warn("Exception from healthcheckHandler.getStatus, setting status to DOWN", e);
        status = InstanceStatus.DOWN;
    }

    if (null != status) {
        // 設置實例狀態，實例狀態變了會觸發狀態變更的監聽器
        applicationInfoManager.setInstanceStatus(status);
    }
}

/////////////////////////////////

public void refreshLeaseInfoIfRequired() {
    // 當前實例續約信息
    LeaseInfo leaseInfo = instanceInfo.getLeaseInfo();
    if (leaseInfo == null) {
        return;
    }
    // 從配置中獲取續約信息
    int currentLeaseDuration = config.getLeaseExpirationDurationInSeconds();
    int currentLeaseRenewal = config.getLeaseRenewalIntervalInSeconds();
    // 如果續約信息變了，就重新創建續約信息，並設置實例為 dirty
    if (leaseInfo.getDurationInSecs() != currentLeaseDuration || leaseInfo.getRenewalIntervalInSecs() != currentLeaseRenewal) {
        LeaseInfo newLeaseInfo = LeaseInfo.Builder.newBuilder()
                .setRenewalIntervalInSecs(currentLeaseRenewal)
                .setDurationInSecs(currentLeaseDuration)
                .build();
        instanceInfo.setLeaseInfo(newLeaseInfo);
        instanceInfo.setIsDirty();
    }
}

/////////////////////////////////

public HealthCheckHandler getHealthCheckHandler() {
    HealthCheckHandler healthCheckHandler = this.healthCheckHandlerRef.get();
    if (healthCheckHandler == null) {
        // 可以自定義 HealthCheckHandler 實現健康檢查
        if (null != healthCheckHandlerProvider) {
            healthCheckHandler = healthCheckHandlerProvider.get();
        } else if (null != healthCheckCallbackProvider) {
            // 可以自定義 HealthCheckCallback 實現健康檢查，HealthCheckCallback 已過期，建議使用 HealthCheckHandler
            healthCheckHandler = new HealthCheckCallbackToHandlerBridge(healthCheckCallbackProvider.get());
        }

        if (null == healthCheckHandler) {
            // 沒有自定義的就是用默認的橋接類
            healthCheckHandler = new HealthCheckCallbackToHandlerBridge(null);
        }
        this.healthCheckHandlerRef.compareAndSet(null, healthCheckHandler);
    }

    return this.healthCheckHandlerRef.get();
}

//////////////////////////////////////

public synchronized void setInstanceStatus(InstanceStatus status) {
    InstanceStatus next = instanceStatusMapper.map(status);
    if (next == null) {
        return;
    }

    // 如果狀態變更了，才會返回之前的狀態，然后觸發狀態變更監聽器
    InstanceStatus prev = instanceInfo.setStatus(next);
    if (prev != null) {
        for (StatusChangeListener listener : listeners.values()) {
            try {
                listener.notify(new StatusChangeEvent(prev, next));
            } catch (Exception e) {
                logger.warn("failed to notify listener: {}", listener.getId(), e);
            }
        }
    }
}

③ 向 eureka server 注冊

在 run 方法里調用了 discoveryClient.register() 方法實現了客戶端實例向注冊中心的注冊，進入到 register 方法可以看到，他就是使用前面構造的 EurekaTransport 來發起遠程調用。

一層層進去，很容易發現就是調用了 eureka-server 的 POST /apps/{appName} 接口，后面我們就從 eureka-core 中找這個接口就可以找到注冊中心實現服務注冊的入口了。

boolean register() throws Throwable {
    logger.info(PREFIX + "{}: registering service...", appPathIdentifier);
    EurekaHttpResponse<Void> httpResponse;
    try {
        // registrationClient => JerseyReplicationClient
        httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
    } catch (Exception e) {
        logger.warn(PREFIX + "{} - registration failed {}", appPathIdentifier, e.getMessage(), e);
        throw e;
    }
    if (logger.isInfoEnabled()) {
        logger.info(PREFIX + "{} - registration status: {}", appPathIdentifier, httpResponse.getStatusCode());
    }
    return httpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode();
}

/////////////////////////////////

public EurekaHttpResponse<Void> register(InstanceInfo info) {
    // 調用的是 POST apps/{appName} 接口
    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 方法
                .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();
        }
    }
}

④ 注冊中心設置實例狀態為已啟動

再回想下注冊中心的初始化流程，在最后調用 openForTraffic 方法時，最后也會調用 ApplicationInfoManager 的 setInstanceStatus 方法，將實例狀態設置為已啟動，這個時候就會觸發客戶端注冊到注冊中心的動作。

applicationInfoManager.setInstanceStatus(InstanceStatus.UP);

⑤ 完成監聽實例變更的方法

狀態變更器會調用 onDemandUpdate 方法來完成實例狀態變更后的邏輯。

• 它這里一個是用到了限流器來限制每分鍾這個方法只能被調用4次，即避免了頻繁的注冊行為
• 然后在調度時，它會從本地變量中取出上一次調度的 Future，如果任務還沒執行完，它會直接取消掉
• 最后就是調用 run 方法，完成服務的注冊

public boolean onDemandUpdate() {
    // 限流控制
    if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) {
        if (!scheduler.isShutdown()) {
            scheduler.submit(new Runnable() {
                @Override
                public void run() {
                    logger.debug("Executing on-demand update of local InstanceInfo");

                    // 如果上一次的任務還沒有執行完，直接取消掉，然后執行注冊的任務
                    Future latestPeriodic = scheduledPeriodicRef.get();
                    if (latestPeriodic != null && !latestPeriodic.isDone()) {
                        logger.debug("Canceling the latest scheduled update, it will be rescheduled at the end of on demand update");
                        latestPeriodic.cancel(false);
                    }

                    InstanceInfoReplicator.this.run();
                }
            });
            return true;
        } else {
            logger.warn("Ignoring onDemand update due to stopped scheduler");
            return false;
        }
    } else {
        logger.warn("Ignoring onDemand update due to rate limiter");
        return false;
    }
}

⑥ 限流器

最后簡單看下限流器 RateLimiter 的設計：

• 從它的注釋中可以看出，eureka 的 RateLimiter 是基於令牌桶算法實現的限流器
• acquire 方法有兩個參數：
  • burstSize：允許以突發方式進入系統的最大請求數
  • averageRate：設置的時間窗口內允許進入的請求數

/**
 * 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>
 * <li>
 *     average rate - expected number of requests per second (RateLimiters using MINUTES is also supported)
 * </li>
 * </ul>
 *
 * @author Tomasz Bak
 */
public class RateLimiter {

    private final long rateToMsConversion;

    private final AtomicInteger consumedTokens = new AtomicInteger();
    private final AtomicLong lastRefillTime = new AtomicLong(0);

    @Deprecated
    public RateLimiter() {
        this(TimeUnit.SECONDS);
    }

    public RateLimiter(TimeUnit averageRateUnit) {
        switch (averageRateUnit) {
            case SECONDS:
                rateToMsConversion = 1000;
                break;
            case MINUTES:
                rateToMsConversion = 60 * 1000;
                break;
            default:
                throw new IllegalArgumentException("TimeUnit of " + averageRateUnit + " is not supported");
        }
    }

    public boolean acquire(int burstSize, long averageRate) {
        return acquire(burstSize, averageRate, System.currentTimeMillis());
    }

    public boolean acquire(int burstSize, long averageRate, long currentTimeMillis) {
        if (burstSize <= 0 || averageRate <= 0) { // Instead of throwing exception, we just let all the traffic go
            return true;
        }

        refillToken(burstSize, averageRate, currentTimeMillis);
        return consumeToken(burstSize);
    }

    private void refillToken(int burstSize, long averageRate, long currentTimeMillis) {
        // 上一次填充 token 的時間
        long refillTime = lastRefillTime.get();
        // 時間差
        long timeDelta = currentTimeMillis - refillTime;
        // 固定生成令牌的速率，即每分鍾4次
        // 例如剛好間隔15秒進來一個請求，就是 15000 * 4 / 60000 = 1，newTokens 代表間隔了多少次，如果等於0，說明間隔不足15秒
        long newTokens = timeDelta * averageRate / rateToMsConversion;
        if (newTokens > 0) {
            long newRefillTime = refillTime == 0
                    ? currentTimeMillis
                    // 注意這里不是直接設置的當前時間戳，而是根據 newTokens 重新計算的，因為有可能同一周期內同時有多個請求進來，這樣可以保持一個固定的周期
                    : refillTime + newTokens * rateToMsConversion / averageRate;
            if (lastRefillTime.compareAndSet(refillTime, newRefillTime)) {
                while (true) {
                    // 調整令牌的數量
                    int currentLevel = consumedTokens.get();
                    int adjustedLevel = Math.min(currentLevel, burstSize);
                    // currentLevel 可能為2，重置為了 0 或 1
                    int newLevel = (int) Math.max(0, adjustedLevel - newTokens);
                    if (consumedTokens.compareAndSet(currentLevel, newLevel)) {
                        return;
                    }
                }
            }
        }
    }

    private boolean consumeToken(int burstSize) {
        while (true) {
            int currentLevel = consumedTokens.get();
            // 突發數量為2，也就是允許15秒內最多有兩次請求進來
            if (currentLevel >= burstSize) {
                return false;
            }
            if (consumedTokens.compareAndSet(currentLevel, currentLevel + 1)) {
                return true;
            }
        }
    }

    public void reset() {
        consumedTokens.set(0);
        lastRefillTime.set(0);
    }
}

3、Eureka Server 接收注冊請求

① 找到實例注冊的API入口

從前面的分析中，我們知道服務端注冊的API是 POST /apps/{appName}，由於 eureka 是基於 jersey 來通信的，想找到API入口還是有點費勁的，至少沒有 springmvc 那么容易。

先看 ApplicationsResource 這個類，可以找到 getApplicationResource 這個方法的路徑是符合 /apps/{appName} 這個規則的。然后可以看到它里面創建了 ApplicationResource，再進入到這個類里面，就可以找到 @Post 標注的 addInstance 方法，這就是注冊的入口了。可以看到它是調用了注冊表的 register 方法來注冊實例的。

@Path("/{version}/apps")
@Produces({"application/xml", "application/json"})
public class ApplicationsResource {
    private final EurekaServerConfig serverConfig;
    private final PeerAwareInstanceRegistry registry;
    private final ResponseCache responseCache;

    // 符合規則 /apps/{appName}
    @Path("{appId}")
    public ApplicationResource getApplicationResource(
            @PathParam("version") String version,
            @PathParam("appId") String appId) {
        CurrentRequestVersion.set(Version.toEnum(version));
        try {
            // 真正的入口
            return new ApplicationResource(appId, serverConfig, registry);
        } finally {
            CurrentRequestVersion.remove();
        }
    }
}

/////////////////////////////////

@Produces({"application/xml", "application/json"})
public class ApplicationResource {

    private final PeerAwareInstanceRegistry registry;

    @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);

        registry.register(info, "true".equals(isReplication));
        return Response.status(204).build();  // 204 to be backwards compatible
    }
}

addInstance 接口有兩個參數：

• InstanceInfo：服務實例，主要有兩塊數據：
  • 基本信息：主機名、IP地址、端口號、URL地址
  • 租約信息：保持心跳的間隔時間、最近心跳的時間、服務注冊的時間、服務啟動的時間
• isReplication：這個參數是從請求頭中取的，表示是否是在同步 server 節點的實例。在集群模式下，因為客戶端實例注冊到注冊中心后，會同步到其它 server節點，所以如果是eureka-server之間同步信息，這個參數就為 true，避免循環同步。

② 實例注冊

進入到注冊表的 register 方法，可以看到主要就是調用父類的 register 方法注冊實例，然后同步到 eureka server 集群中的其它 server 節點。集群同步放到后面來看，現在只需要知道注冊實例時會同步到其它server節點即可。

@Override
public void register(final InstanceInfo info, final boolean isReplication) {
    int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS;
    // 如果實例中沒有周期的配置，就設置為默認的 90 秒
    if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) {
        leaseDuration = info.getLeaseInfo().getDurationInSecs();
    }
    // 注冊實例
    super.register(info, leaseDuration, isReplication);
    // 復制到集群其它 server 節點
    replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication);
}

接着看父類的注冊方法，它的主要流程如下：

• 首先可以看到eureka server保存注冊表（registry）的數據結構是 ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>，key 就是服務名稱，value 就是對應的實例，因為一個服務可能會部署多個實例。
• 根據服務名稱從注冊表拿到實例表，然后根據實例ID拿到實例的租約信息 Lease<InstanceInfo>
• 如果租約信息存在，說明已經注冊過相同的實例了，然后就對比已存在實例和新注冊實例的最后更新時間，如果新注冊的是舊的，就替換為已存在的實例來完成注冊
• 如果租約信息不存在，說明是一個新注冊的實例，這時會更新兩個閾值：
  • 期望續約的客戶端數量 +1
  • 每分鍾續約次數的閾值，如果低於這個值，說明有很多客戶端沒有發送心跳，這時eureka就認為可能網絡出問題了，就會有另一些機制，這個后面再說
• 然后就根據注冊的實例信息和續約周期創建新的租約，並放入注冊表中去
• 接着根據當前時間戳、服務名稱、實例ID封裝一個 Pair，然后放入到最近注冊的隊列中 recentRegisteredQueue，先記住這個隊列就行了
• 根據實例的 overriddenStatus 判斷，不為空的話，可能就只是要更新實例的狀態，這個時候就會只變更實例的狀態，而不會改變 dirty
• 然后是設置了實例的啟動時間戳，設置了實例的 ActionType 為 ADDED
• 將租約加入到最近變更的隊列 recentlyChangedQueue，先記住這個隊列
• 最后一步失效緩存，一步步進去可以發現，主要就是將讀寫緩存 readWriteCacheMap 中與這個實例相關的緩存失效掉，這個緩存后面分析抓取注冊表的時候再來細看

public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
    read.lock();
    try {
        // registry => ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>
        Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
        REGISTER.increment(isReplication);
        if (gMap == null) {
            // 初次注冊時，創建一個 ConcurrentHashMap，key 為 appName
            final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
            gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
            if (gMap == null) {
                gMap = gNewMap;
            }
        }
        Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
        // Retain the last dirty timestamp without overwriting it, if there is already a lease
        if (existingLease != null && (existingLease.getHolder() != null)) {
            // 已存在的實例的最后更新時間
            Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
            // 新注冊的實例的最后更新時間
            Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
            logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);

            // 如果存在的實例比新注冊盡量的實例后更新，就直接把新注冊的實例設置為已存在的實例
            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 {
            // 新注冊時，續約信息不存在
            synchronized (lock) {
                if (this.expectedNumberOfClientsSendingRenews > 0) {
                    // Since the client wants to register it, increase the number of clients sending renews
                    // 期望續約的客戶端數量 + 1
                    this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 1;
                    // 更新每分鍾續約請求次數的閥值，這個閥值在后面很多地方都會用到
                    updateRenewsPerMinThreshold();
                }
            }
            logger.debug("No previous lease information found; it is new registration");
        }
        // 創建新的續約
        Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
        if (existingLease != null) {
            lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
        }
        gMap.put(registrant.getId(), lease);
        // 放入最近注冊的隊列
        recentRegisteredQueue.add(new Pair<Long, String>(
                System.currentTimeMillis(),
                registrant.getAppName() + "(" + registrant.getId() + ")"));
        // 覆蓋狀態
        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);
        // 僅僅是變更實例狀態，不會設置為 dirty
        registrant.setStatusWithoutDirty(overriddenInstanceStatus);

        // If the lease is registered with UP status, set lease service up timestamp
        if (InstanceStatus.UP.equals(registrant.getStatus())) {
            // UP 時設置 Lease 的時間戳
            lease.serviceUp();
        }
        // 設置動作是 ADDED，這個在后面會做 switch 判斷
        registrant.setActionType(ActionType.ADDED);
        // 添加到最近變更的隊列
        recentlyChangedQueue.add(new RecentlyChangedItem(lease));
        // 設置最后更新時間
        registrant.setLastUpdatedTimestamp();
        // 失效緩存
        invalidateCache(registrant.getAppName(), registrant.getVIPAddress(), registrant.getSecureVipAddress());
        logger.info("Registered instance {}/{} with status {} (replication={})",
                registrant.getAppName(), registrant.getId(), registrant.getStatus(), isReplication);
    } finally {
        read.unlock();
    }
}

更新每分鍾續約次數的閾值：

protected void updateRenewsPerMinThreshold() {
    // 每分鍾續約閾值 = 期望續約的客戶端數量 * （60 / 續約間隔時間） * 續約百分比
    // 例如，一共注冊了 10 個實例，那么期望續約的客戶端數量為 10，間隔時間默認為 30秒，就是每個客戶端應該每30秒發送一次心跳，續約百分比默認為 0.85
    // 每分鍾續約次數閾值 = 10 * (60.0 / 30) * 0.85 = 17，也就是說每分鍾至少要接收到 17 此續約請求
    this.numberOfRenewsPerMinThreshold = (int) (this.expectedNumberOfClientsSendingRenews
            * (60.0 / serverConfig.getExpectedClientRenewalIntervalSeconds())
            * serverConfig.getRenewalPercentThreshold());
}

這就是注冊表 registry 緩存服務實例信息的結構，可以看出 eureka 是基於內存來組織注冊表的，使用的是 ConcurrentHashMap 來保證多線程並發安全。

4、Eureka Server 控制台

前面已經將服務實例注冊上去了，現在來看下 eureka server 的控制台頁面是怎么獲取這些數據的。

前面已經分析過 eureka-server 的 web.xml 中配置了歡迎頁為 status.jsp ，這就是控制台的頁面。

從 status.jsp 可以看出，其實就是從 EurekaServerContext 上下文獲取注冊表，然后讀取注冊表注冊的服務實例，然后遍歷展示到表格中。

<%@ page language="java" import="java.util.*,java.util.Map.Entry,com.netflix.discovery.shared.Pair,
com.netflix.discovery.shared.*,com.netflix.eureka.util.*,com.netflix.appinfo.InstanceInfo.*,
com.netflix.appinfo.DataCenterInfo.*,com.netflix.appinfo.AmazonInfo.MetaDataKey,com.netflix.eureka.resources.*,
com.netflix.eureka.*,com.netflix.appinfo.*,com.netflix.eureka.util.StatusUtil" pageEncoding="UTF-8" %>
<%
String path = request.getContextPath();
String basePath = request.getScheme()+"://"+request.getServerName()+":"+request.getServerPort()+path+"/";
%>

<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">

<html>
  <head>
    <base href="<%=basePath%>">

    <title>Eureka</title>
    <link rel="stylesheet" type="text/css" href="./css/main.css">
    <script type="text/javascript" src="./js/jquery-1.11.1.js" ></script>
    <script type="text/javascript" src="./js/jquery.dataTables.js" ></script>
    <script type="text/javascript" >
       $(document).ready(function() {
           $('table.stripeable tr:odd').addClass('odd');
           $('table.stripeable tr:even').addClass('even');
           $('#instances thead th').each(function () {
               var title = $('#instances thead th').eq($(this).index()).text();
               $(this).html(title + '</br><input type="text" placeholder="Search ' + title + '" />');
           });
           // DataTable
           var table = $('#instances').DataTable({"paging": false, "bInfo": false, "sDom": 'ltipr', "bSort": false});
           // Apply the search
           table.columns().eq(0).each(function (colIdx) {
               $('input', table.column(colIdx).header()).on('keyup change', function () {
                   table.column(colIdx).search(this.value).draw();
               });
           });
       });
    </script>
  </head>

  <body id="one">
    <jsp:include page="header.jsp" />
    <jsp:include page="navbar.jsp" />
    <div id="content">
      <div class="sectionTitle">Instances currently registered with Eureka</div>
        <table id='instances' class="stripeable">
           <thead><tr><th>Application</th><th>AMIs</th><th>Availability Zones</th><th>Status</th></tr></thead>
           <tfoot><tr><th>Application</th><th>AMIs</th><th>Availability Zones</th><th>Status</th></tr></tfoot>
           <tbody>
           <%
               // 獲取 eureka server 上下文 EurekaServerContext
           EurekaServerContext serverContext = (EurekaServerContext) pageContext.getServletContext()
                   .getAttribute(EurekaServerContext.class.getName());
           // 從上下文中取出注冊表，
           for(Application app : serverContext.getRegistry().getSortedApplications()) {
               out.print("<tr><td><b>" + app.getName() + "</b></td>");
               Map<String, Integer> amiCounts = new HashMap<String, Integer>();
               Map<InstanceStatus,List<Pair<String, String>>> instancesByStatus =
                   new HashMap<InstanceStatus, List<Pair<String,String>>>();
               Map<String,Integer> zoneCounts = new HashMap<String, Integer>();

               for(InstanceInfo info : app.getInstances()){
                   String id = info.getId();
                   String url = info.getStatusPageUrl();
                   InstanceStatus status = info.getStatus();
                   String ami = "n/a";
                   String zone = "";
                   if(info.getDataCenterInfo().getName() == Name.Amazon){
                       AmazonInfo dcInfo = (AmazonInfo)info.getDataCenterInfo();
                       ami = dcInfo.get(MetaDataKey.amiId);
                       zone = dcInfo.get(MetaDataKey.availabilityZone);
                   }

                   Integer count = amiCounts.get(ami);
                   if(count != null){
                       amiCounts.put(ami, Integer.valueOf(count.intValue()+1));
                   }else {
                       amiCounts.put(ami, Integer.valueOf(1));
                   }

                   count = zoneCounts.get(zone);
                   if(count != null){
                       zoneCounts.put(zone, Integer.valueOf(count.intValue()+1));
                   }else {
                       zoneCounts.put(zone, Integer.valueOf(1));
                   }
                   List<Pair<String, String>> list = instancesByStatus.get(status);

                   if(list == null){
                       list = new ArrayList<Pair<String,String>>();
                       instancesByStatus.put(status, list);
                   }
                   list.add(new Pair<String, String>(id, url));
               }
               StringBuilder buf = new StringBuilder();
               for (Iterator<Entry<String, Integer>> iter =
                   amiCounts.entrySet().iterator(); iter.hasNext();) {
                   Entry<String, Integer> entry = iter.next();
                   buf.append("<b>").append(entry.getKey()).append("</b> (").append(entry.getValue()).append("), ");
               }
               out.println("<td>" + buf.toString() + "</td>");
               buf = new StringBuilder();
               for (Iterator<Entry<String, Integer>> iter =
                   zoneCounts.entrySet().iterator(); iter.hasNext();) {
                   Entry<String, Integer> entry = iter.next();
                   buf.append("<b>").append(entry.getKey()).append("</b> (").append(entry.getValue()).append("), ");
               }
               out.println("<td>" + buf.toString() + "</td>");
               buf = new StringBuilder();
               for (Iterator<Entry<InstanceStatus, List<Pair<String,String>>>> iter =
                   instancesByStatus.entrySet().iterator(); iter.hasNext();) {
                   Entry<InstanceStatus, List<Pair<String,String>>> entry = iter.next();
                   List<Pair<String, String>> value = entry.getValue();
                   InstanceStatus status = entry.getKey();
                   if(status != InstanceStatus.UP){
                       buf.append("<font color=red size=+1><b>");
                   }
                   buf.append("<b>").append(status.name()).append("</b> (").append(value.size()).append(") - ");
                   if(status != InstanceStatus.UP){
                       buf.append("</font></b>");
                   }

                   for(Pair<String,String> p : value) {
                       String id = p.first();
                       String url = p.second();
                       if(url != null && url.startsWith("http")){
                           buf.append("<a href=\"").append(url).append("\">");
                       }else {
                           url = null;
                       }
                       buf.append(id);
                       if(url != null){
                           buf.append("</a>");
                       }
                       buf.append(", ");
                   }
               }
               out.println("<td>" + buf.toString() + "</td></tr>");
           }
           %>
           </tbody>
           </table>
      </div>
      <div>
      <div class="sectionTitle">General Info</div>
      <table id='generalInfo' class="stripeable">
          <tr><th>Name</th><th>Value</th></tr>
           <%
           StatusInfo statusInfo = (new StatusUtil(serverContext)).getStatusInfo();
           Map<String,String> genMap = statusInfo.getGeneralStats();
           for (Map.Entry<String,String> entry : genMap.entrySet()) {
             out.print("<tr>");
             out.print("<td>" + entry.getKey() +  "</td><td>" + entry.getValue() + "</td>");
             out.print("</tr>");
           }
           Map<String,String> appMap = statusInfo.getApplicationStats();
           for (Map.Entry<String,String> entry : appMap.entrySet()) {
             out.print("<tr>");
             out.print("<td>" + entry.getKey() +  "</td><td>" + entry.getValue() + "</td>");
             out.print("</tr>");
           }
           %>
           </table>
      </div>
      <div>
      <div class="sectionTitle">Instance Info</div>
        <table id='instanceInfo' class="stripeable">
          <tr><th>Name</th><th>Value</th></tr>
           <%
           InstanceInfo instanceInfo = statusInfo.getInstanceInfo();
           Map<String,String> instanceMap = new HashMap<String,String>();
           instanceMap.put("ipAddr", instanceInfo.getIPAddr());
           instanceMap.put("status", instanceInfo.getStatus().toString());
           if(instanceInfo.getDataCenterInfo().getName() == DataCenterInfo.Name.Amazon) {
               AmazonInfo info = (AmazonInfo) instanceInfo.getDataCenterInfo();
               instanceMap.put("availability-zone", info.get(AmazonInfo.MetaDataKey.availabilityZone));
               instanceMap.put("public-ipv4", info.get(AmazonInfo.MetaDataKey.publicIpv4));
               instanceMap.put("instance-id", info.get(AmazonInfo.MetaDataKey.instanceId));
               instanceMap.put("public-hostname", info.get(AmazonInfo.MetaDataKey.publicHostname));
               instanceMap.put("ami-id", info.get(AmazonInfo.MetaDataKey.amiId));
               instanceMap.put("instance-type", info.get(AmazonInfo.MetaDataKey.instanceType));
           }
           for (Map.Entry<String,String> entry : instanceMap.entrySet()) {
             out.print("<tr>");
             out.print("<td>" + entry.getKey() +  "</td><td>" + entry.getValue() + "</td>");
             out.print("</tr>");
           }
           %>
           </table>
    </div>

  </body>
</html>

5、服務注冊的整體流程圖

下面通過一張圖來看看服務實例注冊的整個流程。

六、抓取注冊表

1、Eureka Client 啟動時全量抓取注冊表

客戶端啟動初始化 DiscoveryClient 時，其中有段代碼如下：這一步調用 fetchRegistry 就是在啟動時全量抓取注冊表緩存到本地中。

if (clientConfig.shouldFetchRegistry()) {
    try {
        // 拉取注冊表：全量抓取和增量抓取
        boolean primaryFetchRegistryResult = fetchRegistry(false);
        if (!primaryFetchRegistryResult) {
            logger.info("Initial registry fetch from primary servers failed");
        }
        boolean backupFetchRegistryResult = true;
        if (!primaryFetchRegistryResult && !fetchRegistryFromBackup()) {
            backupFetchRegistryResult = false;
            logger.info("Initial registry fetch from backup servers failed");
        }
        if (!primaryFetchRegistryResult && !backupFetchRegistryResult && clientConfig.shouldEnforceFetchRegistryAtInit()) {
            throw new IllegalStateException("Fetch registry error at startup. Initial fetch failed.");
        }
    } catch (Throwable th) {
        logger.error("Fetch registry error at startup: {}", th.getMessage());
        throw new IllegalStateException(th);
    }
}

進入 fetchRegistry 方法，可以看到，首先獲取本地的 Applications，如果為空就會調用 getAndStoreFullRegistry 方法全量抓取注冊表並緩存到本地。

private boolean fetchRegistry(boolean forceFullRegistryFetch) {
    Stopwatch tracer = FETCH_REGISTRY_TIMER.start();

    try {
        // 獲取本地的應用實例
        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
        {
            // 全量抓取注冊表
            getAndStoreFullRegistry();
        } else {
            // 增量更新注冊表
            getAndUpdateDelta(applications);
        }
        applications.setAppsHashCode(applications.getReconcileHashCode());
        logTotalInstances();
    } catch (Throwable e) {
        logger.info(PREFIX + "{} - was unable to refresh its cache! This periodic background refresh will be retried in {} seconds. status = {} stacktrace = {}",
                appPathIdentifier, clientConfig.getRegistryFetchIntervalSeconds(), e.getMessage(), ExceptionUtils.getStackTrace(e));
        return false;
    } finally {
        if (tracer != null) {
            tracer.stop();
        }
    }

    // 發出緩存刷新的通知
    onCacheRefreshed();

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

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

進入 getAndStoreFullRegistry  方法可以發現，就是調用 GET /apps 接口抓取全量注冊表，因此等會服務端就從這個入口進去看抓取全量注冊表的邏輯。注冊表抓取回來之后，就放到本地變量 localRegionApps 中。localRegionApps 的類型是 AtomicReference<Applications>，實例信息是存儲在 Applications 中的。

private void getAndStoreFullRegistry() throws Throwable {
    long currentUpdateGeneration = fetchRegistryGeneration.get();

    logger.info("Getting all instance registry info from the eureka server");

    Applications apps = null;
    EurekaHttpResponse<Applications> httpResponse = clientConfig.getRegistryRefreshSingleVipAddress() == null
            // 調用 server GET /apps 全量抓取注冊表
            ? eurekaTransport.queryClient.getApplications(remoteRegionsRef.get())
            : eurekaTransport.queryClient.getVip(clientConfig.getRegistryRefreshSingleVipAddress(), remoteRegionsRef.get());
    if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
        apps = httpResponse.getEntity();
    }
    logger.info("The response status is {}", httpResponse.getStatusCode());

    if (apps == null) {
        logger.error("The application is null for some reason. Not storing this information");
    } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
        // 將 Applications 緩存到本地
        localRegionApps.set(this.filterAndShuffle(apps));
        logger.debug("Got full registry with apps hashcode {}", apps.getAppsHashCode());
    } else {
        logger.warn("Not updating applications as another thread is updating it already");
    }
}

2、Eureka Server 注冊表多級緩存機制

① 全量抓取注冊表的接口

全量抓取注冊表的接口是 GET /apps，跟找注冊接口是類似的，最終可以找到 ApplicationsResource 的 getContainers 方法就是全量抓取注冊表的入口。

• 可以看出，我們可以通過請求頭來指定返回 xml 格式還是 json 格式，可以指定是否要壓縮返回等。
• 然后創建了全量緩存的 Key
• 接着根據緩存的 key 從 responseCache 中全量抓取注冊表

@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) {
    // 省略部分代碼...

    // JSON 類型
    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
    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 {
        // 根據緩存 key 從 responseCache 獲取全量注冊表
        response = Response.ok(responseCache.get(cacheKey))
                .build();
    }
    CurrentRequestVersion.remove();
    return response;
}

② ResponseCache 多級緩存讀取

ResponseCache 就是 eureka server 讀取注冊表的核心組件，它的內部采用了多級緩存的機制來快速響應客戶端抓取注冊表的請求，下面就來看看 ResponseCache。

緩存讀取的流程：

• 如果設置了使用只讀緩存（默認true），就先從只讀緩存 readOnlyCacheMap 中讀取；readOnlyCacheMap 使用 ConcurrentHashMap 實現，ConcurrentHashMap 支持並發訪問，讀取速度很快。
• 如果讀寫緩存中沒有，就從讀寫緩存 readWriteCacheMap 中讀取，讀取出來后並寫入到只讀緩存中；readWriteCacheMap 使用 google guava 的 LoadingCache 實現，LoadingCache 支持在沒有元素的時候使用 CacheLoader 加載元素。
• 如果沒有開啟使用只讀緩存，就直接從讀寫緩存中獲取。

public String get(final Key key) {
    return get(key, shouldUseReadOnlyResponseCache);
}

////////////////////////////////////////////////////

String get(final Key key, boolean useReadOnlyCache) {
    // => getValue
    Value payload = getValue(key, useReadOnlyCache);
    if (payload == null || payload.getPayload().equals(EMPTY_PAYLOAD)) {
        return null;
    } else {
        return payload.getPayload();
    }
}

////////////////////////////////////////////////////

Value getValue(final Key key, boolean useReadOnlyCache) {
    Value payload = null;
    try {
        if (useReadOnlyCache) {
            // 開啟使用只讀緩存，則先從只讀緩存讀取
            // readOnlyCacheMap => ConcurrentHashMap<Key, Value>
            final Value currentPayload = readOnlyCacheMap.get(key);
            if (currentPayload != null) {
                payload = currentPayload;
            } else {
                // 只讀緩存中沒有，則從讀寫緩存中讀取，然后放入只讀緩存中
                // readWriteCacheMap => LoadingCache<Key, Value>
                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;
}

③ ResponseCache 初始化

分析 eureka server EurekaBootStrap 啟動初始化時，最后有一步去初始化 eureka server 上下文，它里面就會去初始化注冊表，初始化注冊表的時候就會初始化 ResponseCache，這里就來分析下這個初始化干了什么。

• 主要就是使用 google guava cache 構造了一個讀寫緩存 readWriteCacheMap，初始容量為 1000。注意這個讀寫緩存的特性：每隔 180 秒定時過期，然后元素不存在的時候就會使用 CacheLoader 從注冊表中讀取。
• 接着如果配置了使用只讀緩存，還會開啟一個定時任務，每隔30秒將讀寫緩存 readWriteCacheMap 的數據同步到只讀緩存 readOnlyCacheMap。

ResponseCacheImpl(EurekaServerConfig serverConfig, ServerCodecs serverCodecs, AbstractInstanceRegistry registry) {
    this.serverConfig = serverConfig;
    this.serverCodecs = serverCodecs;
    // 是否使用只讀緩存，默認為 true
    this.shouldUseReadOnlyResponseCache = serverConfig.shouldUseReadOnlyResponseCache();
    // 保存注冊表
    this.registry = registry;
    // 緩存更新間隔時間，默認30秒
    long responseCacheUpdateIntervalMs = serverConfig.getResponseCacheUpdateIntervalMs();
    // 使用 google guava cache 構造一個讀寫緩存
    this.readWriteCacheMap =
            // 初始容量為1000
            CacheBuilder.newBuilder().initialCapacity(serverConfig.getInitialCapacityOfResponseCache())
                    // 緩存的數據在寫入多久后過期，默認180秒，也就是說 readWriteCacheMap 會定時過期
                    .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對應的元素不存在時，使用定義 CacheLoader 加載元素
                    .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;
                        }
                    });

    if (shouldUseReadOnlyResponseCache) {
        // 如果配置了使用只讀緩存，就開啟一個定時任務，定期將 readWriteCacheMap 的數據同步到 readOnlyCacheMap 中
        // 默認間隔時間是 30 秒
        timer.schedule(getCacheUpdateTask(),
                new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
                        + responseCacheUpdateIntervalMs),
                responseCacheUpdateIntervalMs);
    }

    try {
        Monitors.registerObject(this);
    } catch (Throwable e) {
        logger.warn("Cannot register the JMX monitor for the InstanceRegistry", e);
    }
}

generatePayload 方法：

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: {} found in the cache key.", key.getEntityType());
                payload = "";
                break;
        }
        return new Value(payload);
    } finally {
        if (tracer != null) {
            tracer.stop();
        }
    }
}

3、Eureka Server 注冊表多級緩存過期機制

這節來總結下 eureka server 注冊表多級緩存的過期時機，其實前面都已經分析過了。

① 主動過期

分析服務注冊時已經說過，服務注冊完成后，調用了 invalidateCache 來失效緩存，進去可以看到就是將讀寫緩存 readWriteCacheMap 中的服務、所有服務、增量服務的緩存失效掉。

那這里就要注意了，如果服務注冊、下線、故障之類的，這里只是失效了讀寫緩存，然后可能要間隔30秒才能同步到只讀緩存 readOnlyCacheMap，那么其它客戶端可能要隔30秒后才能感知到。

private void invalidateCache(String appName, @Nullable String vipAddress, @Nullable String secureVipAddress) {
    // invalidate cache
    responseCache.invalidate(appName, vipAddress, secureVipAddress);
}

緩存失效：

@Override
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),
                    // 失效所有 APP 的緩存
                    new Key(Key.EntityType.Application, ALL_APPS, type, v, EurekaAccept.full),
                    new Key(Key.EntityType.Application, ALL_APPS, type, v, EurekaAccept.compact),
                    // 失效增量 APP 的緩存
                    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 在構建的時候，指定了一個自動過期的時間，默認值是180秒，所以往 readWriteCacheMap 中放入一個數據過后，等180秒過后，它就自動過期了。然后下次讀取的時候發現緩存中沒有這個 key，就會使用 CacheLoader 重新加載到這個緩存中。

這種定時過期機制就是每隔一段時間來同步注冊表與緩存的數據。

③ 被動過期

初始化 ResponseCache 時，如果啟用了只讀緩存，就會創建一個定時任務（每隔30秒運行一次）來同步 readWriteCacheMap 與 readOnlyCacheMap 中的數據，對於 readOnlyCacheMap 來說這就是一種被動過期。

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()) {
                    logger.debug("Updating the client cache from response cache for key : {} {} {} {}",
                            key.getEntityType(), key.getName(), key.getVersion(), key.getType());
                }
                try {
                    CurrentRequestVersion.set(key.getVersion());
                    // 獲取讀寫緩存中的數據
                    Value cacheValue = readWriteCacheMap.get(key);
                    // 獲取只讀緩存中的數據
                    Value currentCacheValue = readOnlyCacheMap.get(key);
                    // 如果 readOnlyCacheMap 中緩存的值與 readWriteCacheMap 緩存的值不同，就用 readWriteCacheMap 的值覆蓋 readOnlyCacheMap 的值
                    if (cacheValue != currentCacheValue) {
                        readOnlyCacheMap.put(key, cacheValue);
                    }
                } catch (Throwable th) {
                    logger.error("Error while updating the client cache from response cache for key {}", key.toStringCompact(), th);
                } finally {
                    CurrentRequestVersion.remove();
                }
            }
        }
    };
}

4、Eureka Client 定時拉取增量注冊表

① 客戶端注冊表刷新定時任務

前面介紹 DiscoveryClient 初始化時，在初始化調度任務這一步，如果要抓取注冊表，就會創建一個調度器每隔 30 秒執行一次 cacheRefreshTask，它對 CacheRefreshThread 做了封裝，進去可以看到，它其實就是調用 refreshRegistry 方法刷新注冊表。

private void initScheduledTasks() {
    if (clientConfig.shouldFetchRegistry()) {
        // 抓取注冊表的間隔時間，默認30秒
        int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
        // 刷新緩存調度器延遲時間擴大倍數，在任務超時的時候，將擴大延遲時間
        // 這在出現網絡抖動、eureka-sever 不可用時，可以避免頻繁發起無效的調度
        int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
        // 注冊表刷新的定時任務
        cacheRefreshTask = new TimedSupervisorTask(
                "cacheRefresh",
                scheduler,
                cacheRefreshExecutor,
                registryFetchIntervalSeconds,
                TimeUnit.SECONDS,
                expBackOffBound,
                new CacheRefreshThread() // 刷新注冊表的任務
        );
        // 30秒后開始調度刷新注冊表的任務
        scheduler.schedule(
                cacheRefreshTask,
                registryFetchIntervalSeconds, TimeUnit.SECONDS);
    }
}

refreshRegistry 方法：

class CacheRefreshThread implements Runnable {
    public void run() {
        refreshRegistry();
    }
}

@VisibleForTesting
void refreshRegistry() {
    try {
        boolean isFetchingRemoteRegionRegistries = isFetchingRemoteRegionRegistries();

        boolean remoteRegionsModified = false;
        // This makes sure that a dynamic change to remote regions to fetch is honored.
        String latestRemoteRegions = clientConfig.fetchRegistryForRemoteRegions();
        if (null != latestRemoteRegions) {
            String currentRemoteRegions = remoteRegionsToFetch.get();
            if (!latestRemoteRegions.equals(currentRemoteRegions)) {
                // Both remoteRegionsToFetch and AzToRegionMapper.regionsToFetch need to be in sync
                synchronized (instanceRegionChecker.getAzToRegionMapper()) {
                    if (remoteRegionsToFetch.compareAndSet(currentRemoteRegions, latestRemoteRegions)) {
                        String[] remoteRegions = latestRemoteRegions.split(",");
                        remoteRegionsRef.set(remoteRegions);
                        instanceRegionChecker.getAzToRegionMapper().setRegionsToFetch(remoteRegions);
                        remoteRegionsModified = true;
                    } else {
                        logger.info("Remote regions to fetch modified concurrently," +
                                " ignoring change from {} to {}", currentRemoteRegions, latestRemoteRegions);
                    }
                }
            } else {
                // Just refresh mapping to reflect any DNS/Property change
                instanceRegionChecker.getAzToRegionMapper().refreshMapping();
            }
        }

        // 抓取注冊表
        boolean success = fetchRegistry(remoteRegionsModified);
        if (success) {
            registrySize = localRegionApps.get().size();
            lastSuccessfulRegistryFetchTimestamp = System.currentTimeMillis();
        }

        if (logger.isDebugEnabled()) {
            StringBuilder allAppsHashCodes = new StringBuilder();
            allAppsHashCodes.append("Local region apps hashcode: ");
            allAppsHashCodes.append(localRegionApps.get().getAppsHashCode());
            allAppsHashCodes.append(", is fetching remote regions? ");
            allAppsHashCodes.append(isFetchingRemoteRegionRegistries);
            for (Map.Entry<String, Applications> entry : remoteRegionVsApps.entrySet()) {
                allAppsHashCodes.append(", Remote region: ");
                allAppsHashCodes.append(entry.getKey());
                allAppsHashCodes.append(" , apps hashcode: ");
                allAppsHashCodes.append(entry.getValue().getAppsHashCode());
            }
            logger.debug("Completed cache refresh task for discovery. All Apps hash code is {} ",
                    allAppsHashCodes);
        }
    } catch (Throwable e) {
        logger.error("Cannot fetch registry from server", e);
    }
}

refreshRegistry 里面又調用了 fetchRegistry 抓取注冊表，fetchRegistry 在前面分析全量抓取注冊表時已經展示過了。全量抓取注冊表之后，本地 applications 不為空了，這時就會走 getAndUpdateDelta 增量更新的方法。

private boolean fetchRegistry(boolean forceFullRegistryFetch) {
    Stopwatch tracer = FETCH_REGISTRY_TIMER.start();

    try {
        // 獲取本地的應用實例
        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.info(PREFIX + "{} - was unable to refresh its cache! This periodic background refresh will be retried in {} seconds. status = {} stacktrace = {}",
                appPathIdentifier, clientConfig.getRegistryFetchIntervalSeconds(), e.getMessage(), ExceptionUtils.getStackTrace(e));
        return false;
    } finally {
        if (tracer != null) {
            tracer.stop();
        }
    }

    // 發出刷新緩存的通知
    onCacheRefreshed();

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

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

② 增量更新本地注冊表

接着看 getAndUpdateDelta 增量更新方法：

• 首先調用 eureka server GET /apps/delta 接口獲取增量的注冊表
• 如果增量的注冊表為空，就會調用 getAndStoreFullRegistry 方法全量抓取注冊表
• 增量注冊表不為空，就將其合並到本地注冊表中
• 然后根據本地注冊表的 applications 重新計算一個 hash 值
• eureka server 返回的 delta 中包含一個 appsHashCode，代表了 eureka server 端的注冊表的 hash 值，如果與本地計算的 hash 值不同，則說明本地注冊表與server端注冊表不一致，那就會全量拉取注冊表更新到本地緩存中

可以看到，eureka 增量抓取的思路來更新本地緩存，並使用了 hash 值來保證服務端與本地的數據一致性。在分布式系統里，要進行數據同步，采用 hash 值比對的思想，這是值得學習的一個思路。

private void getAndUpdateDelta(Applications applications) throws Throwable {
    long currentUpdateGeneration = fetchRegistryGeneration.get();

    Applications delta = null;
    // 調用遠程接口增量抓取：GET apps/delta
    EurekaHttpResponse<Applications> httpResponse = eurekaTransport.queryClient.getDelta(remoteRegionsRef.get());
    if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
        delta = httpResponse.getEntity();
    }

    // 如果增量抓取的數據為空，就會進行一次全量抓取
    if (delta == null) {
        logger.warn("The server does not allow the delta revision to be applied because it is not safe. "
                + "Hence got the full registry.");
        getAndStoreFullRegistry();
    }

    else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
        logger.debug("Got delta update with apps hashcode {}", delta.getAppsHashCode());
        String reconcileHashCode = "";
        // 加鎖更新本地注冊表
        if (fetchRegistryUpdateLock.tryLock()) {
            try {
                // 抓取到增量的注冊表后，跟本地的注冊表合並
                updateDelta(delta);
                // 注冊表合並完成后，根據本地 applications 計算一個 hash 值
                reconcileHashCode = getReconcileHashCode(applications);
            } finally {
                fetchRegistryUpdateLock.unlock();
            }
        } else {
            logger.warn("Cannot acquire update lock, aborting getAndUpdateDelta");
        }
        // delta 中會返回 server 端注冊表的 hash 值，如果和本地計算出來的 hash 值不一樣，
        // 說明本地注冊表跟 server 端注冊表不一樣，就會從 server 全量拉取注冊表更新到本地緩存
        if (!reconcileHashCode.equals(delta.getAppsHashCode()) || clientConfig.shouldLogDeltaDiff()) {
            reconcileAndLogDifference(delta, reconcileHashCode);  // this makes a remoteCall
        }
    } else {
        logger.warn("Not updating application delta as another thread is updating it already");
        logger.debug("Ignoring delta update with apps hashcode {}, as another thread is updating it already", delta.getAppsHashCode());
    }
}

③ 增量注冊表合並到本地

再來看下增量注冊表合並到本地發方法 updateDelta，其實就是遍歷返回來的服務實例，然后根據實例的  ActionType 分別處理，比如前面分析實例注冊時 ActionType 就設置了 ADDED，后面分析實例下線時還可以看到設置了 ActionType 為 DELETED。

private void updateDelta(Applications delta) {
    int deltaCount = 0;
    // 變量增量注冊的服務
    for (Application app : delta.getRegisteredApplications()) {
        // 遍歷實例
        for (InstanceInfo instance : app.getInstances()) {
            Applications applications = getApplications();
            String instanceRegion = instanceRegionChecker.getInstanceRegion(instance);
            if (!instanceRegionChecker.isLocalRegion(instanceRegion)) {
                Applications remoteApps = remoteRegionVsApps.get(instanceRegion);
                if (null == remoteApps) {
                    remoteApps = new Applications();
                    remoteRegionVsApps.put(instanceRegion, remoteApps);
                }
                applications = remoteApps;
            }

            ++deltaCount;
            // ADDED 新增的實例：服務注冊
            if (ActionType.ADDED.equals(instance.getActionType())) {
                Application existingApp = applications.getRegisteredApplications(instance.getAppName());
                if (existingApp == null) {
                    applications.addApplication(app);
                }
                logger.debug("Added instance {} to the existing apps in region {}", instance.getId(), instanceRegion);
                applications.getRegisteredApplications(instance.getAppName()).addInstance(instance);
            }
            // MODIFIED 變更的實例：續約，信息變更
            else if (ActionType.MODIFIED.equals(instance.getActionType())) {
                Application existingApp = applications.getRegisteredApplications(instance.getAppName());
                if (existingApp == null) {
                    applications.addApplication(app);
                }
                logger.debug("Modified instance {} to the existing apps ", instance.getId());

                applications.getRegisteredApplications(instance.getAppName()).addInstance(instance);

            }
            // DELETED 移除實例：實例下線、故障
            else if (ActionType.DELETED.equals(instance.getActionType())) {
                Application existingApp = applications.getRegisteredApplications(instance.getAppName());
                if (existingApp != null) {
                    logger.debug("Deleted instance {} to the existing apps ", instance.getId());
                    existingApp.removeInstance(instance);
                    /*
                     * 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.
                     */
                    if (existingApp.getInstancesAsIsFromEureka().isEmpty()) {
                        applications.removeApplication(existingApp);
                    }
                }
            }
        }
    }
    logger.debug("The total number of instances fetched by the delta processor : {}", deltaCount);

    getApplications().setVersion(delta.getVersion());
    getApplications().shuffleInstances(clientConfig.shouldFilterOnlyUpInstances());

    for (Applications applications : remoteRegionVsApps.values()) {
        applications.setVersion(delta.getVersion());
        applications.shuffleInstances(clientConfig.shouldFilterOnlyUpInstances());
    }
}

④ 客戶端實例存儲

可以看到，全量更新或者增量更新都是在更新本地的 Applications，可以看下 Applications 和 Application 的結構：

• 應用實例是被封裝到 Application 中的，實例列表是一個 LinkedHashSet，並用一個 ConcurrentHashMap 緩存了實例ID和實例的關系，便於快速檢索。
• 應用 Application 再被封裝到 Applications 中，應用列表是一個 ConcurrentLinkedQueue，並用 ConcurrentHashMap 緩存了應用名稱和 Application 的關系。
• Applications 中還保存了這個應用的 appsHashCode  值，這個應該是和 eureka server 的 hash 是一致的。

private final AbstractQueue<Application> applications;
private final Map<String, Application> appNameApplicationMap;
private final Map<String, VipIndexSupport> virtualHostNameAppMap;
private final Map<String, VipIndexSupport> secureVirtualHostNameAppMap;

public Applications(@JsonProperty("appsHashCode") String appsHashCode,
        @JsonProperty("versionDelta") Long versionDelta,
        @JsonProperty("application") List<Application> registeredApplications) {
    // 實例應用隊列
    this.applications = new ConcurrentLinkedQueue<Application>();
    // key 為 appName，Value 為對應的應用
    this.appNameApplicationMap = new ConcurrentHashMap<String, Application>();
    this.virtualHostNameAppMap = new ConcurrentHashMap<String, VipIndexSupport>();
    this.secureVirtualHostNameAppMap = new ConcurrentHashMap<String, VipIndexSupport>();
    // hash 值
    this.appsHashCode = appsHashCode;
    this.versionDelta = versionDelta;

    for (Application app : registeredApplications) {
        this.addApplication(app);
    }
}

public class Application {

    private static Random shuffleRandom = new Random();

    private String name;

    // 是否變更
    private volatile boolean isDirty = false;
    // 實例信息
    private final Set<InstanceInfo> instances;

    private final AtomicReference<List<InstanceInfo>> shuffledInstances;
    // map 結構實例
    private final Map<String, InstanceInfo> instancesMap;

    public Application() {
        instances = new LinkedHashSet<InstanceInfo>();
        instancesMap = new ConcurrentHashMap<String, InstanceInfo>();
        shuffledInstances = new AtomicReference<List<InstanceInfo>>();
    }

}

5、Eureka Server 返回增量注冊表

① 抓取增量注冊表的入口

從前分析知道，增量抓取注冊表單接口為 GET/apps/delta，可以很容易找到位於 ApplicationsResource 下的 getContainerDifferential 就是抓取增量注冊表的入口。

可以看到，跟抓取注冊表類似，也是先構建一個緩存的Key，然后從多級緩存 ResponseCache 中獲取。這里的key是 ALL_APPS_DELTA。

@Path("delta")
@GET
public Response getContainerDifferential(
        @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) {

    Key cacheKey = new Key(Key.EntityType.Application,
            // 增量服務：ALL_APPS_DELTA
            ResponseCacheImpl.ALL_APPS_DELTA,
            keyType, CurrentRequestVersion.get(), EurekaAccept.fromString(eurekaAccept), regions
    );

    final 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();
    }

    CurrentRequestVersion.remove();
    return response;
}

與全量抓取注冊表，讀取多級緩存的流程都是類似的，唯一的區別就是 Key 不同，全量抓取時是 ALL_APPS，增量抓取時 ALL_APPS_DELTA，區別就在於 readWriteCacheMap 加載數據到緩存中時走的邏輯不一樣，可以再看看下面的 generatePayload 方法就知道了。

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: {} found in the cache key.", key.getEntityType());
                payload = "";
                break;
        }
        return new Value(payload);
    } finally {
        if (tracer != null) {
            tracer.stop();
        }
    }
}

② 增量注冊表的設計

之后會調用 registry.getApplicationDeltas() 獲取增量注冊表，進去可以發現，增量的注冊表其實就是 recentlyChangedQueue 這個最近變更隊列里的數據，通過遍歷 recentlyChangedQueue 生成 Applications。

在返回 apps 之前，先獲取了本地所有應用，並計算了一個 hash 值，然后設置到 apps 中。這就和前一節對應起來了，抓取增量注冊表時，服務端會返回一個全量注冊表的 hash 值，然后客戶端將增量注冊表合並到本地后，再根據本地的全量注冊表計算一個 hash 值，然后將兩個 hash 值做對比，如果不一致，說明服務端和客戶端的數據是不一致的，這時客戶端就會重新向服務端全量拉取注冊表到本地。

public Applications getApplicationDeltas() {
    GET_ALL_CACHE_MISS_DELTA.increment();
    Applications apps = new Applications();
    apps.setVersion(responseCache.getVersionDelta().get());
    Map<String, Application> applicationInstancesMap = new HashMap<String, Application>();
    write.lock();
    try {
        // 最近變更隊列 recentlyChangedQueue，這就是增量的注冊表
        // recentlyChangedQueue 只保留了最近3分鍾有變化的實例，如實例上線、下線、故障剔除
        Iterator<RecentlyChangedItem> iter = this.recentlyChangedQueue.iterator();
        logger.debug("The number of elements in the delta queue is : {}",
                this.recentlyChangedQueue.size());
        while (iter.hasNext()) {
            Lease<InstanceInfo> lease = iter.next().getLeaseInfo();
            InstanceInfo instanceInfo = lease.getHolder();
            logger.debug(
                    "The instance id {} is found with status {} and actiontype {}",
                    instanceInfo.getId(), instanceInfo.getStatus().name(), instanceInfo.getActionType().name());
            Application app = applicationInstancesMap.get(instanceInfo
                    .getAppName());
            if (app == null) {
                app = new Application(instanceInfo.getAppName());
                applicationInstancesMap.put(instanceInfo.getAppName(), app);
                apps.addApplication(app);
            }
            app.addInstance(new InstanceInfo(decorateInstanceInfo(lease)));
        }

        // 省略部分代碼...

        // 獲取所有應用實例
        Applications allApps = getApplications(!disableTransparentFallback);
        // 根據所有應用實例計算一個 hash 值，並設置到要返回的 apps 中
        apps.setAppsHashCode(allApps.getReconcileHashCode());
        return apps;
    } finally {
        write.unlock();
    }
}

再來看看 recentlyChangedQueue 是如何設計來保存增量信息的。

再看看前面提到過的注冊表初始化的構造方法，最后創建了一個每隔30秒執行一次的定時調度任務。這個任務會遍歷 recentlyChangedQueue 這個隊列，判斷每個元素的最后更新時間是否超過了 180 秒，如果超過了，就會從隊列中移除這個元素。超過 180 秒的實例變更信息，就會認為這些變更信息都已經同步到客戶端了，因為客戶端是每隔30秒拉取一次增量注冊表的。因此客戶端多次拉取增量注冊表可能拉取到同樣的變更信息，不過最終合並到本地都是一樣的。

因此可以看出，eureka 利用 recentlyChangedQueue 這個最近變更隊列保存了最近3分鍾以內實例的變更信息，如新服務注冊、服務下線等，然后客戶端每次就是拉取這個變更隊列。

protected AbstractInstanceRegistry(EurekaServerConfig serverConfig, EurekaClientConfig clientConfig, ServerCodecs serverCodecs) {
    this.serverConfig = serverConfig;
    this.clientConfig = clientConfig;
    this.serverCodecs = serverCodecs;
    // 最近下線的循環隊列
    this.recentCanceledQueue = new CircularQueue<Pair<Long, String>>(1000);
    // 最近注冊的循環隊列
    this.recentRegisteredQueue = new CircularQueue<Pair<Long, String>>(1000);

    // 最近一分鍾續約的計數器
    this.renewsLastMin = new MeasuredRate(1000 * 60 * 1);

    // 一個定時調度任務，定時剔除最近改變隊列中過期的實例
    this.deltaRetentionTimer.schedule(getDeltaRetentionTask(),
            // 調度任務延遲 30 秒開始執行
            serverConfig.getDeltaRetentionTimerIntervalInMs(),
            // 默認每隔 30 秒執行一次
            serverConfig.getDeltaRetentionTimerIntervalInMs());
}

/////////////////////////////////////////

private TimerTask getDeltaRetentionTask() {
    return new TimerTask() {

        @Override
        public void run() {
            // 最近變更的隊列
            Iterator<RecentlyChangedItem> it = recentlyChangedQueue.iterator();
            while (it.hasNext()) {
                // 最近更新時間超過 180 秒就認為數據已經同步到各個客戶端了，就從隊列中移除
                if (it.next().getLastUpdateTime() <
                        // retentionTimeInMSInDeltaQueue：delta隊列數據保留時間，默認 180 秒
                        System.currentTimeMillis() - serverConfig.getRetentionTimeInMSInDeltaQueue()) {
                    it.remove();
                } else {
                    break;
                }
            }
        }

    };
}

7、Eureka 抓取注冊表總體流程圖

下面還是用一張圖整體展示下服務抓取注冊表的整理流程。

服務注冊、服務下線、實例故障剔除都會將讀寫緩存 readWriteCacheMap 中對應的實例失效掉，然后加入到最近變更隊列 recentlyChangedQueue 中，因此這三種情況下，增量抓取注冊表的邏輯都是類似的。

七、服務續約

在分布式系統中，服務續約機制是非常重要的，這樣能讓中心系統（注冊中心）知道客戶端還存活着。接下來就來看看服務續約的機制。

1、Eureka Client 定時發送心跳

在初始化 DiscoveryClient 的調度任務時，下面這部分代碼就是在創建定時發送心跳的任務，心跳每隔30秒發送一次。發送心跳的接口是 PUT /apps/{appName}/{instanceId}。

private void initScheduledTasks() {
    // 定時刷新本地緩存...

    if (clientConfig.shouldRegisterWithEureka()) {
        // 續約間隔時間，默認30秒
        int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
        // 心跳調度器的延遲時間擴大倍數，默認10
        int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
        logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs);

        // 心跳的定時任務
        heartbeatTask = new TimedSupervisorTask(
                "heartbeat",
                scheduler,
                heartbeatExecutor,
                renewalIntervalInSecs,
                TimeUnit.SECONDS,
                expBackOffBound,
                new HeartbeatThread()
        );
        // 30秒后開始調度心跳的任務
        scheduler.schedule(
                heartbeatTask,
                renewalIntervalInSecs, TimeUnit.SECONDS);

        // 服務注冊...
    } else {
        logger.info("Not registering with Eureka server per configuration");
    }
}

//////////////////////////////////////////////

private class HeartbeatThread implements Runnable {
    public void run() {
        if (renew()) {
            lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
        }
    }
}

//////////////////////////////////////////////

boolean renew() {
    EurekaHttpResponse<InstanceInfo> httpResponse;
    try {
        // 發送心跳的接口：PUT /apps/{appName}/{instanceId}
        httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
        logger.debug(PREFIX + "{} - Heartbeat status: {}", appPathIdentifier, httpResponse.getStatusCode());
        if (httpResponse.getStatusCode() == Status.NOT_FOUND.getStatusCode()) {
            REREGISTER_COUNTER.increment();
            logger.info(PREFIX + "{} - Re-registering apps/{}", appPathIdentifier, instanceInfo.getAppName());
            long timestamp = instanceInfo.setIsDirtyWithTime();
            // 服務端未找到對應的實例，就重新注冊
            boolean success = register();
            if (success) {
                instanceInfo.unsetIsDirty(timestamp);
            }
            return success;
        }
        // 續約成功
        return httpResponse.getStatusCode() == Status.OK.getStatusCode();
    } catch (Throwable e) {
        logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e);
        return false;
    }
}

2、Eureka Server 接收心跳續約

順着 PUT /apps/{appName}/{instanceId} 找可以發現，服務端接收注冊的入口在 InstanceResource 的 renewLease 方法中，它調用了注冊表單 renew 方法進行服務續約。

@PUT
public Response renewLease(
        @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication,
        @QueryParam("overriddenstatus") String overriddenStatus,
        @QueryParam("status") String status,
        @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
    if (!isSuccess) {
        logger.warn("Not Found (Renew): {} - {}", app.getName(), id);
        return 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
    Response response;
    if (lastDirtyTimestamp != null && serverConfig.shouldSyncWhenTimestampDiffers()) {
        response = this.validateDirtyTimestamp(Long.valueOf(lastDirtyTimestamp), isFromReplicaNode);
        // Store the overridden status since the validation found out the node that replicates wins
        if (response.getStatus() == Response.Status.NOT_FOUND.getStatusCode()
                && (overriddenStatus != null)
                && !(InstanceStatus.UNKNOWN.name().equals(overriddenStatus))
                && isFromReplicaNode) {
            registry.storeOverriddenStatusIfRequired(app.getAppName(), id, InstanceStatus.valueOf(overriddenStatus));
        }
    } else {
        response = Response.ok().build();
    }
    logger.debug("Found (Renew): {} - {}; reply status={}", app.getName(), id, response.getStatus());
    return response;
}

進去可以看到，調用了父類的 renew 方法續約，然后會判斷 isReplication ，如果是復制，說明是來自 eureka-server 集群中其它節點的同步請求，就復制到其它節點。復制到其它集群這塊代碼在前面已經提到過了，就不再展示。

public boolean renew(final String appName, final String id, final boolean isReplication) {
    // 調用父類（AbstractInstanceRegistry）的 renew 續約
    if (super.renew(appName, id, isReplication)) {
        // 續約完成后同步到集群其它節點
        replicateToPeers(Action.Heartbeat, appName, id, null, null, isReplication);
        return true;
    }
    return false;
}

接着看父類的 renew 續約方法：

• 首先根據服務名從注冊表中取出租約信息
• 然后根據實例ID取出實例的租約信息
• 然后判斷是否是覆蓋實例狀態
• 將最近一分鍾續約次數計數器 renewsLastMin +1
• 最后調用實例租約對象的 renew 方法進行續約，其內部只是更新了租約的最后更新時間 lastUpdateTimestamp ，更新為當前時間+續約間隔時間。

public boolean renew(String appName, String id, boolean isReplication) {
    RENEW.increment(isReplication);
    // 根據服務名從注冊表取出租約信息
    Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
    Lease<InstanceInfo> leaseToRenew = null;
    if (gMap != null) {
        // 根據實例ID取出實例租約信息
        leaseToRenew = gMap.get(id);
    }
    if (leaseToRenew == null) {
        RENEW_NOT_FOUND.increment(isReplication);
        logger.warn("DS: Registry: lease doesn't exist, registering resource: {} - {}", appName, id);
        return false;
    } else {
        InstanceInfo instanceInfo = leaseToRenew.getHolder();
        if (instanceInfo != null) {
            // touchASGCache(instanceInfo.getASGName());
            InstanceStatus overriddenInstanceStatus = this.getOverriddenInstanceStatus(
                    instanceInfo, leaseToRenew, isReplication);
            if (overriddenInstanceStatus == InstanceStatus.UNKNOWN) {
                logger.info("Instance status UNKNOWN possibly due to deleted override for instance {}"
                        + "; re-register required", instanceInfo.getId());
                RENEW_NOT_FOUND.increment(isReplication);
                return false;
            }
            if (!instanceInfo.getStatus().equals(overriddenInstanceStatus)) {
                logger.info(
                        "The instance status {} is different from overridden instance status {} for instance {}. "
                                + "Hence setting the status to overridden status", instanceInfo.getStatus().name(),
                                overriddenInstanceStatus.name(),
                                instanceInfo.getId());
                instanceInfo.setStatusWithoutDirty(overriddenInstanceStatus);

            }
        }
        // 最近一分鍾續約計數器+1
        renewsLastMin.increment();
        // 續約
        leaseToRenew.renew();
        return true;
    }
}

////////////////////////////////////////////////

public void renew() {
    // 更新最后更新時間，在當前時間的基礎上加了周期時間，默認90秒
    lastUpdateTimestamp = System.currentTimeMillis() + duration;
}

八、服務下線

1、Eureka Client 下線

eureka client 服務關閉停止時，會觸發 DiscoveryClient 的 shutdown 關閉 eureka-client，我們就從 shutdown 方法來看看 eureka-client 的下線。

• 首先移除了注冊的狀態變更器，這個時候不再需要監聽實例狀態的變更了
• 然后關閉了一系列的調度任務，停止與 eureka-server 的交互，比如定時發送心跳。同時也釋放了資源。
• 之后調用了 unregister 取消注冊，其實就是調用了 server 端的 DELETE /apps/{appName}/{instanceId} 下線實例
• 最后再關閉了一些其它資源，如 EurekaTransport。

@PreDestroy
@Override
public synchronized void shutdown() {
    if (isShutdown.compareAndSet(false, true)) {
        logger.info("Shutting down DiscoveryClient ...");

        // 移除狀態變更監聽器
        if (statusChangeListener != null && applicationInfoManager != null) {
            applicationInfoManager.unregisterStatusChangeListener(statusChangeListener.getId());
        }

        // 停止調度任務，釋放資源：
        //    instanceInfoReplicator、heartbeatExecutor、cacheRefreshExecutor
        //    scheduler、cacheRefreshTask、heartbeatTask
        cancelScheduledTasks();

        // If APPINFO was registered
        if (applicationInfoManager != null
                && clientConfig.shouldRegisterWithEureka()
                && clientConfig.shouldUnregisterOnShutdown()) {
            applicationInfoManager.setInstanceStatus(InstanceStatus.DOWN);
            // 調用 eureka-server 的下線接口下線實例
            unregister();
        }

        // 繼續釋放資源
        if (eurekaTransport != null) {
            eurekaTransport.shutdown();
        }
        heartbeatStalenessMonitor.shutdown();
        registryStalenessMonitor.shutdown();

        Monitors.unregisterObject(this);

        logger.info("Completed shut down of DiscoveryClient");
    }
}

void unregister() {
    // It can be null if shouldRegisterWithEureka == false
    if(eurekaTransport != null && eurekaTransport.registrationClient != null) {
        try {
            logger.info("Unregistering ...");
            // 取消注冊：DELETE /apps/{appName}/{instanceId}
            EurekaHttpResponse<Void> httpResponse = eurekaTransport.registrationClient.cancel(instanceInfo.getAppName(), instanceInfo.getId());
            logger.info(PREFIX + "{} - deregister  status: {}", appPathIdentifier, httpResponse.getStatusCode());
        } catch (Exception e) {
            logger.error(PREFIX + "{} - de-registration failed{}", appPathIdentifier, e.getMessage(), e);
        }
    }
}

2、Eureka Server 服務下線

順着 DELETE /apps/{appName}/{instanceId} 接口可以找到 InstanceResouce 的 cancelLease 方法就是客戶端下線的入口。

進入注冊的 cancel 方法，可以看到與前面的一些接口是類似的，先調用服務的 cancel 方法下線實例，然后調用 replicateToPeers 復制到集群中其它節點。然后 cancel 方法其實是調用的 internalCancel 方法。

@DELETE
public Response cancelLease(@HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) {
    try {
        // 下線實例
        boolean isSuccess = registry.cancel(app.getName(), id,
            "true".equals(isReplication));

        if (isSuccess) {
            logger.debug("Found (Cancel): {} - {}", app.getName(), id);
            return Response.ok().build();
        } else {
            logger.info("Not Found (Cancel): {} - {}", app.getName(), id);
            return Response.status(Status.NOT_FOUND).build();
        }
    } catch (Throwable e) {
        logger.error("Error (cancel): {} - {}", app.getName(), id, e);
        return Response.serverError().build();
    }
}

//////////////////////////////////////////

public boolean cancel(final String appName, final String id,
                      final boolean isReplication) {
    if (super.cancel(appName, id, isReplication)) {
        replicateToPeers(Action.Cancel, appName, id, null, null, isReplication);

        return true;
    }
    return false;
}

//////////////////////////////////////////

public boolean cancel(String appName, String id, boolean isReplication) {
    return internalCancel(appName, id, isReplication);
}

再來看下 internalCancel 方法：

• 首先根據服務名從注冊表取出服務所有實例的租約信息，再根據實例ID移除實例租約信息
• 將移除的實例加入到最近下線的一個循環隊列 recentCanceledQueue
• 下線實例，注意這里設置了實例的下線時間 evictionTimestamp 為當前時間
• 然后設置實例的 ActionType 為 DELETED，再將下線的實例加入最近變更的隊列 recentlyChangedQueue
• 之后失效掉緩存 readWriteCacheMap，服務實例變更了就必須清理緩存。不過  readWriteCacheMap 可能要30秒才會同步到 readOnlyCacheMap。
• 最后將期望續約的客戶端數量-1，然后更新了每分鍾續約次數閾值

protected boolean internalCancel(String appName, String id, boolean isReplication) {
    read.lock();
    try {
        CANCEL.increment(isReplication);
        // 根據服務名稱取出服務租約信息
        Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
        Lease<InstanceInfo> leaseToCancel = null;
        if (gMap != null) {
            // 根據實例ID移除實例租約信息
            leaseToCancel = gMap.remove(id);
        }
        // 將移除的實例ID加入到最近下線的隊列中
        recentCanceledQueue.add(new Pair<Long, String>(System.currentTimeMillis(), appName + "(" + id + ")"));
        InstanceStatus instanceStatus = overriddenInstanceStatusMap.remove(id);
        if (instanceStatus != null) {
            logger.debug("Removed instance id {} from the overridden map which has value {}", id, instanceStatus.name());
        }
        if (leaseToCancel == null) {
            CANCEL_NOT_FOUND.increment(isReplication);
            logger.warn("DS: Registry: cancel failed because Lease is not registered for: {}/{}", appName, id);
            return false;
        } else {
            // 下線實例，設置了實例的下線時間 evictionTimestamp 為當前時間戳
            leaseToCancel.cancel();
            InstanceInfo instanceInfo = leaseToCancel.getHolder();
            String vip = null;
            String svip = null;
            if (instanceInfo != null) {
                // 設置實例 ActionType 為 DELETED
                instanceInfo.setActionType(ActionType.DELETED);
                // 加入最近變更隊列中
                recentlyChangedQueue.add(new RecentlyChangedItem(leaseToCancel));
                instanceInfo.setLastUpdatedTimestamp();
                vip = instanceInfo.getVIPAddress();
                svip = instanceInfo.getSecureVipAddress();
            }
            // 失效緩存
            invalidateCache(appName, vip, svip);
            logger.info("Cancelled instance {}/{} (replication={})", appName, id, isReplication);
        }
    } finally {
        read.unlock();
    }

    synchronized (lock) {
        // 期望續約的客戶端數量 - 1
        if (this.expectedNumberOfClientsSendingRenews > 0) {
            // Since the client wants to cancel it, reduce the number of clients to send renews.
            this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews - 1;
            // 更新每分鍾續約次數的閾值
            updateRenewsPerMinThreshold();
        }
    }

    return true;
}

九、服務故障

服務正常停止時，會觸發 DiscoveryClient 的 shutdown 方法關閉 eureka-client，並向注冊中心發送下線的通知。但如果客戶端宕機或非正常關機，注冊中心就無法接收到客戶端下線的通知了，這時注冊中心就會有一個定時任務，根據心跳來判斷客戶端實例是否故障下線了，然后摘除故障的實例。

1、摘除實例的定時任務初始化

在 EurekaBootStrap 初始化的最后幾步中，調用了注冊表的 openForTraffic 做一些最后的設置，其中最后一步調用了 super.postInit 方法做最后的初始化，里面就創建了定時摘除過期實例的調度任務。

registry.openForTraffic(applicationInfoManager, registryCount);

public void openForTraffic(ApplicationInfoManager applicationInfoManager, int count) {
    // 期望的客戶端每分鍾的續約次數
    this.expectedNumberOfClientsSendingRenews = count;
    // 更新每分鍾續約閾值
    updateRenewsPerMinThreshold();
    logger.info("Got {} instances from neighboring DS node", count);
    logger.info("Renew threshold is: {}", numberOfRenewsPerMinThreshold);
    this.startupTime = System.currentTimeMillis();
    if (count > 0) {
        this.peerInstancesTransferEmptyOnStartup = false;
    }
    DataCenterInfo.Name selfName = applicationInfoManager.getInfo().getDataCenterInfo().getName();
    boolean isAws = Name.Amazon == selfName;
    if (isAws && serverConfig.shouldPrimeAwsReplicaConnections()) {
        logger.info("Priming AWS connections for all replicas..");
        primeAwsReplicas(applicationInfoManager);
    }
    logger.info("Changing status to UP");
    // 設置實例狀態為已啟動
    applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
    // 調用父類的后置初始化
    super.postInit();
}

postInit：

• 首先是開啟了最近一分鍾續約次數的計數器
• 然后創建了定時摘除過期實例的調度任務，調度任務每隔60秒執行一次

protected void postInit() {
    // 啟動 統計最近一分鍾續約次數的計數器
    renewsLastMin.start();
    if (evictionTaskRef.get() != null) {
        evictionTaskRef.get().cancel();
    }
    // 定時剔除任務
    evictionTaskRef.set(new EvictionTask());
    evictionTimer.schedule(evictionTaskRef.get(),
            serverConfig.getEvictionIntervalTimerInMs(),
            // 每隔60秒執行一次
            serverConfig.getEvictionIntervalTimerInMs());
}

2、定時摘除過期的實例

① 摘除實例的定時任務

可以看到，每次運行摘除實例的任務時，會先獲取一個補償時間，因為兩次 EvictionTask 執行的間隔時間可能超過了設置的60秒，比如 GC 導致的停頓或本地時間漂移導致計時不准確等。然后就調用了 evict 方法摘除實例。

在計算時間差的場景中，這種補償時間的思路是值得學習的，要考慮到時間差的不准確性。

class EvictionTask extends TimerTask {

    private final AtomicLong lastExecutionNanosRef = new AtomicLong(0l);

    @Override
    public void run() {
        try {
            // 獲取補償時間，因為兩次 EvictionTask 執行的間隔時間可能超過了設置的60秒，比如 GC 導致的停頓或本地時間漂移導致計時不准確
            long compensationTimeMs = getCompensationTimeMs();
            logger.info("Running the evict task with compensationTime {}ms", compensationTimeMs);
            evict(compensationTimeMs);
        } catch (Throwable e) {
            logger.error("Could not run the evict task", e);
        }
    }

    long getCompensationTimeMs() {
        long currNanos = getCurrentTimeNano();
        long lastNanos = lastExecutionNanosRef.getAndSet(currNanos);
        if (lastNanos == 0L) {
            return 0L;
        }
        // 兩次任務運行的間隔時間
        long elapsedMs = TimeUnit.NANOSECONDS.toMillis(currNanos - lastNanos);
        // 補償時間 = 任務運行間隔時間 - 剔除任務的間隔時間（默認60秒）
        long compensationTime = elapsedMs - serverConfig.getEvictionIntervalTimerInMs();
        return compensationTime <= 0L ? 0L : compensationTime;
    }

    long getCurrentTimeNano() {  // for testing
        return System.nanoTime();
    }
}

② 摘除實例

摘除實例的過程如下：

• 首先判斷是否啟用了租約過期的機制（主要就是自我保護機制，下一章節再說）。
• 遍歷注冊表，判斷實例是否過期，將過期的實例加入集合列表中。
• 計算摘除實例的數量限制，主要就是出於自我保護機制，避免一次摘除過多實例。
• 然后就是從要摘除的集合中隨機選擇限制數量內的過期實例來摘除掉。
• 摘除實例其實就是調用了實例下線的方法 internalCancel，主要就是從注冊表中移除實例、加入最近變更的隊列、失效緩存等，具體可以回看服務下線那節。

public void evict(long additionalLeaseMs) {
    logger.debug("Running the evict task");

    // 是否啟用了租約過期
    if (!isLeaseExpirationEnabled()) {
        logger.debug("DS: lease expiration is currently disabled.");
        return;
    }

    // We collect first all expired items, to evict them in random order. For large eviction sets,
    // if we do not that, we might wipe out whole apps before self preservation kicks in. By randomizing it,
    // the impact should be evenly distributed across all applications.
    List<Lease<InstanceInfo>> expiredLeases = new ArrayList<>();
    for (Entry<String, Map<String, Lease<InstanceInfo>>> groupEntry : registry.entrySet()) {
        Map<String, Lease<InstanceInfo>> leaseMap = groupEntry.getValue();
        if (leaseMap != null) {
            for (Entry<String, Lease<InstanceInfo>> leaseEntry : leaseMap.entrySet()) {
                Lease<InstanceInfo> lease = leaseEntry.getValue();
                // 判斷實例租約是否過期
                if (lease.isExpired(additionalLeaseMs) && lease.getHolder() != null) {
                    // 加入到過期的集合列表中
                    expiredLeases.add(lease);
                }
            }
        }
    }

    // 先獲取注冊表已注冊的實例數量
    int registrySize = (int) getLocalRegistrySize();
    // 注冊表數量保留的閾值：已注冊的實例數 * 續約百分比閾值（默認0.85）
    int registrySizeThreshold = (int) (registrySize * serverConfig.getRenewalPercentThreshold());
    // 剔除的數量限制，也就是說一次最多只能剔除 15% 的實例
    int evictionLimit = registrySize - registrySizeThreshold;

    // 得到最小的剔除數量
    int toEvict = Math.min(expiredLeases.size(), evictionLimit);
    if (toEvict > 0) {
        logger.info("Evicting {} items (expired={}, evictionLimit={})", toEvict, expiredLeases.size(), evictionLimit);

        Random random = new Random(System.currentTimeMillis());
        for (int i = 0; i < toEvict; i++) {
            // 根據要剔除的數量從 expiredLeases 中隨機剔除 toEvict 個過期實例
            int next = i + random.nextInt(expiredLeases.size() - i);
            Collections.swap(expiredLeases, i, next);
            Lease<InstanceInfo> lease = expiredLeases.get(i);

            String appName = lease.getHolder().getAppName();
            // 實例ID
            String id = lease.getHolder().getId();
            EXPIRED.increment();
            logger.warn("DS: Registry: expired lease for {}/{}", appName, id);
            // 調用下線的方法
            internalCancel(appName, id, false);
        }
    }
}

③ 分批次摘取實例

可以看到，過期的實例並不是一次性摘除的，而是計算了一個閾值 toEvict，一次只隨機摘除 toEvict 個過期實例，采用了分批摘取+隨機摘取的機制。

比如注冊表一共有20個實例，那么最多可以摘除的實例數 toEvict = 20 - 20 * 0.85 = 3 個，也就是說就算有5個實例過期了，那這一次也只能隨機摘除其中3個，另外兩個要等到下一次執行摘除任務時再摘除。

這種分批摘取機制+隨機摘取機制可能會導致有些過期實例要過很久才能下線，尤其是在開發環境這種頻繁啟動、停止服務的場景中。

3、如何判斷實例是否過期

從上面可以看到，eureka 調用了 lease.isExpired(additionalLeaseMs) 來判斷實例是否過期。進入 isExpired 這個方法可以看到，如果已經設置了摘除時間，或者 當前時間 > （實例最后更新時間 + 續約周期(90秒) + 補償時間），就認為實例已經過期了，說明實例已經超過一個周期沒有續約了，就認為這個客戶端實例發生了故障，無法續約，要被摘除掉。

/**
 * Checks if the lease of a given {@link com.netflix.appinfo.InstanceInfo} has expired or not.
 *
 * Note that due to renew() doing the 'wrong" thing and setting lastUpdateTimestamp to +duration more than
 * what it should be, the expiry will actually be 2 * duration. This is a minor bug and should only affect
 * instances that ungracefully shutdown. Due to possible wide ranging impact to existing usage, this will
 * not be fixed.
 *
 * @param additionalLeaseMs any additional lease time to add to the lease evaluation in ms.
 */
public boolean isExpired(long additionalLeaseMs) {
    // 已經設置過剔除時間，或者 當前時間 > （實例最后更新時間 + 續約周期(90秒) + 補償時間）
    return (evictionTimestamp > 0 || System.currentTimeMillis() > (lastUpdateTimestamp + duration + additionalLeaseMs));
}

這里其實要注意的是另外一個問題，可以看看 isExpired 的注釋，eureka 說這其實是一個bug，但不打算修復了，因為它的 duration 其實是被加了兩次的，下面來看看怎么回事。

先看下 lastUpdateTimestamp 是如何更新的，在客戶端續約的時候會更新 lastUpdateTimestamp，將其設置為 當前時間 + duration 間隔周期（默認90秒），

public void renew() {
    // 更新最后更新時間，在當前時間的基礎上加了一個周期間隔時間，默認90秒
    lastUpdateTimestamp = System.currentTimeMillis() + duration;
}

這個 duration 在注冊的時候也有設置，我們通過這個來看看它的含義是什么。可以看到，如果客戶端沒有配置 durationInSecs，就會設置為默認的 90 秒。

從 getDurationInSecs 的注釋可以看出，這個 duration 的意思是等待客戶端多久沒有續約之后就將其剔除，默認為 90 秒。比如客戶端每隔 30 秒續約一次，那可能超過3次沒有續約之后，就會認為客戶端實例故障了，就要將其摘除掉。

public void register(final InstanceInfo info, final boolean isReplication) {
    int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS;
    // 如果實例中沒有周期的配置，就設置為默認的 90 秒
    if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) {
        leaseDuration = info.getLeaseInfo().getDurationInSecs();
    }
    // 注冊實例
    super.register(info, leaseDuration, isReplication);
    // 復制到集群其它 server 節點
    replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication);
}

//////////////////////////////////////////////

/**
 * Returns client specified setting for eviction (e.g. how long to wait w/o
 * renewal event)
 *
 * @return time in milliseconds since epoch.
 */
public int getDurationInSecs() {
    return durationInSecs;
}

但實際上並不是90秒后摘除實例，可以看到 isExpired 里面將 lastUpdateTimestamp 又加了一個 duration，也就是 180 秒了。也就是說客戶端實例超過 180 秒未續約才被認為是故障了，然后要將其摘除。

isExpired 的注釋也說了，續約的方法 renew() 錯誤的計算了 lastUpdateTimestamp，實際的過期周期是 2 * duration，但是 eureka 並不打算修復這個bug，因為它的影響范圍很小。

所以這里得出一個結論，客戶端關閉了（非正常下線），注冊表中的實例並不是90秒后就摘除了，至少是 180 秒后才會被摘除。

十、自我保護機制

如果網段偶爾抖動或暫時不可用，就無法接收到客戶端的續約，因此 eureka server 為了保證可用性，就會去判斷最近一分鍾收到的心跳次數是否小於指定的閾值，是的就會觸發自我保護機制，關閉租約失效剔除，不再摘除實例，從而保護注冊信息。

1、摘除實例時的自我保護機制

摘除實例的 evict 方法調用了 isLeaseExpirationEnabled 這個方法判斷是否觸發自我保護機制，如果返回 false，就不會摘除實例了。

先看看 isLeaseExpirationEnabled 這個方法：

• 首先，如果沒有啟用自我保護機制，就返回 true，那就可以摘除實例
• 如果啟用了自我保護機制（默認啟用），就判斷每分鍾續約閾值 > 0 且 最近一分鍾續約次數 > 每分鍾續約閾值 就是啟用了租約過期

public boolean isLeaseExpirationEnabled() {
    // 先判斷是否啟用了自我保護機制
    if (!isSelfPreservationModeEnabled()) {
        // The self preservation mode is disabled, hence allowing the instances to expire.
        return true;
    }
    // 每分鍾續約閾值大於0, 且 最近一分鍾續約次數 大於 每分鍾續約閾值
    return numberOfRenewsPerMinThreshold > 0 && getNumOfRenewsInLastMin() > numberOfRenewsPerMinThreshold;
}

public boolean isSelfPreservationModeEnabled() {
    return serverConfig.shouldEnableSelfPreservation();
}

這個每分鍾續約閾值 numberOfRenewsPerMinThreshold 在前面很多地方都看到過了，服務注冊、下線、openForTraffic、以及有個定時任務每隔15分鍾都會調用下面這個方法來更新 numberOfRenewsPerMinThreshold。

protected void updateRenewsPerMinThreshold() {
    // 每分鍾續約閾值 = 期望續約的客戶端數量 * （60 / 續約間隔時間） * 續約閾值
    // 例如，一共注冊了 10 個實例，那么期望續約的客戶端數量為 10，間隔時間默認為 30秒，就是每個客戶端應該每30秒發送一次心跳，續約百分比默認為 0.85
    // 每分鍾續約次數閾值 = 10 * (60.0 / 30) * 0.85 = 17，也就是說每分鍾至少要接收到 17 此續約請求
    this.numberOfRenewsPerMinThreshold = (int) (this.expectedNumberOfClientsSendingRenews
            * (60.0 / serverConfig.getExpectedClientRenewalIntervalSeconds())
            * serverConfig.getRenewalPercentThreshold());
}

expectedNumberOfClientsSendingRenews 在實例注冊的時候會 + 1，在實例下線的時候會 - 1，其代表的就是期望續約的客戶端數量。

/////////////// 實例注冊 ///////////////
synchronized (lock) {
    if (this.expectedNumberOfClientsSendingRenews > 0) {
        // Since the client wants to register it, increase the number of clients sending renews
        // 期望續約的客戶端數量 + 1
        this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews + 1;
        // 更新每分鍾續約請求次數的閾值，這個閾值在后面很多地方都會用到
        updateRenewsPerMinThreshold();
    }
}


/////////////// 實例下線（下線、故障摘除） ///////////////
synchronized (lock) {
    // 期望續約的客戶端數量 - 1
    if (this.expectedNumberOfClientsSendingRenews > 0) {
        // Since the client wants to cancel it, reduce the number of clients to send renews.
        this.expectedNumberOfClientsSendingRenews = this.expectedNumberOfClientsSendingRenews - 1;
        // 更新每分鍾續約次數的閾值
        updateRenewsPerMinThreshold();
    }
}

而最近一分鍾續約次數計數器 renewsLastMin 在每個客戶端續約的時候就會+1，可以回看下 renew 方法，最后調用了 renewsLastMin.increment() 增加一次續約次數。而 renewsLastMin.getCount() 返回的是上一分鍾總的續約次數。

public long getNumOfRenewsInLastMin() {
    return renewsLastMin.getCount();
}

根據以上代碼舉個例子來看看實例故障時的自我保護機制：

• 比如注冊了20個實例，實例默認發送心跳續約的間隔時間為30秒，續約的閾值為 0.85，並且開啟了自我保護機制。
• 那么期望續約的客戶端數量 expectedNumberOfClientsSendingRenews = 20，每分鍾發送心跳的閾值 numberOfRenewsPerMinThreshold = 20 * （60 / 30 ）* 0.85 = 34。
• 正常來說20個實例每分鍾發送心跳的次數 renewsLastMin = 20 * （60 / 30）= 40 次。
• 那么 numberOfRenewsPerMinThreshold（34） > 0 && renewsLastMin（40）> numberOfRenewsPerMinThreshold（34）就是滿足的，就允許摘除故障的實例。
• 那如果有 3 個實例上一分鍾沒有發送續約了，這個時候 renewsLastMin = 17 * （60 / 30）= 34 次，而 numberOfRenewsPerMinThreshold 還是不變，因為注冊表的實例並未移除，因此這時條件就不滿足了，就算實例真的故障了，也不能摘除實例了。

這就是 eureka-server 的自我保護機制，他認為如果短時間內有過的的實例未發送心跳（超過15%），它會認為是自己網絡故障導致客戶端不能發送心跳，就進入自我保護機制，避免誤摘除實例。

2、自我保護機制導致實例未下線的情況

在開發環境中，因為會頻繁重啟服務，會發現有些服務已經是下線狀態了（DOWN），但服務實例一直未被摘除，就是因為 eureka-server 的自我保護機制導致的，下面來看下。

① 啟用自我保護機制的情況

首先 eureka-server 做了如下配置，啟用注冊中心：

eureka:
  server:
    # 是否啟用自我保護機制
    enable-self-preservation: true

啟動幾個客戶端實例：

然后快速將 demo-consumer 停止掉（如果正常關閉，會調用 cancel 下線實例），這時就會看到 demo-consumer 已經DOWN了，但是實例一直未被移除。

可以看到，上一分鍾續約的次數為 4 次，期望每分鍾續約次數為6次，因為不滿足判斷的條件，所以就觸發了自我保護機制，導致一直無法摘除實例。

注意期望續約的客戶端數量為4，而實際注冊的客戶端實例是3個，這是因為 springcloud 在調用 openForTraffic 設置了初始值為 1。

② 關閉自我保護機制的情況

配置如下，關閉自我保護機制：

eureka:
  server:
    # 是否啟用自我保護機制
    enable-self-preservation: false

這時注冊中心控制台會提示我們關閉了自我保護機制：

同樣的操作，快速停掉實例，發現實例還是未被摘除：

那其實是因為實例要180秒后才會被認為是過期的，所以等3分鍾以后，實例就會下線了。

public boolean isExpired(long additionalLeaseMs) {
    return (evictionTimestamp > 0 || System.currentTimeMillis() > (lastUpdateTimestamp + duration + additionalLeaseMs));
}

③ 快速關閉多個實例

這次同時關閉2個實例來看看，在2分鍾之后，發現只有一個實例下線了，這因為eureka-server一次只會摘除15%的實例。

④ DOWN 是怎么來的

那么DOWN這個狀態是怎么來的呢？由於我本地是用IDEA啟動的客戶端實例，其實在關閉的時候，會觸發狀態變更監聽器，然后就會觸發一次注冊的調用，注冊的狀態是 DOWN，因此實例狀態馬上就變為 DOWN 了。

如果直接 kill 這個進程，就不會觸發狀態變更監聽器了，注冊中心的實例就不會變為 DOWN 了，但是實例已經下線變為不可用的狀態了。

⑤ 實例快速下線

經過前面的測試可以總結出來，要想實例快速下線，可以調整如下一些參數。

eureka-server 配置：

eureka:
  server:
    # 是否啟用自我保護機制
    enable-self-preservation: false
    # 每分鍾續約閾值
    renewal-percent-threshold: 0
    # 摘除實例的定時任務的間隔時間
    eviction-interval-timer-in-ms: 10000

eureka-client 配置：

eureka:
  instance:
    # 判斷實例多久未發送心跳就判定為過期
    lease-expiration-duration-in-seconds: 60

3、最近一分鍾計數器的設計

來看下最近一分鍾續約次數計數器 renewsLastMin 是如何統計上一分鍾的續約次數的，renewsLastMin 的類型是 MeasuredRate，這個類的設計也是值得學習的一個點。

MeasuredRate 利用兩個桶來計數，一個保存上一間隔時間的計數，一個保存當前這一間隔時間的計數，然后使用定時任務每隔一定間隔時間就將當前這個桶的計數替換到上一個桶里。然后增加計數的時候增加當前桶，獲取數量的時候從上一個桶里獲取，就實現了獲取上一個間隔時間的計數。

public class MeasuredRate {
    private static final Logger logger = LoggerFactory.getLogger(MeasuredRate.class);
    // 利用了兩個桶來計數，一個是上一分鍾，一個是當前這一分鍾
    private final AtomicLong lastBucket = new AtomicLong(0);
    private final AtomicLong currentBucket = new AtomicLong(0);

    private final long sampleInterval;
    private final Timer timer;

    private volatile boolean isActive;

    /**
     * @param sampleInterval in milliseconds
     */
    public MeasuredRate(long sampleInterval) {
        // 間隔時間
        this.sampleInterval = sampleInterval;
        // 定時器
        this.timer = new Timer("Eureka-MeasureRateTimer", true);
        this.isActive = false;
    }

    public synchronized void start() {
        if (!isActive) {
            timer.schedule(new TimerTask() {
                @Override
                public void run() {
                    try {
                        // 每分鍾執行一次，將當前這一分鍾的次數設置到上一分鍾的桶里
                        lastBucket.set(currentBucket.getAndSet(0));
                    } catch (Throwable e) {
                        logger.error("Cannot reset the Measured Rate", e);
                    }
                }
            }, sampleInterval, sampleInterval);

            isActive = true;
        }
    }

    public synchronized void stop() {
        if (isActive) {
            timer.cancel();
            isActive = false;
        }
    }

    /**
     * Returns the count in the last sample interval.
     */
    public long getCount() {
        // 獲取計數時是獲取的上一分鍾這個桶的計數
        return lastBucket.get();
    }

    /**
     * Increments the count in the current sample interval.
     */
    public void increment() {
        // 增加計數的時候是增加的當前這個桶的計數
        currentBucket.incrementAndGet();
    }
}

4、服務故障摘除和自我保護機制圖

下面用一張圖來總結下服務故障摘除和自我保護機制。