前言

前情回顧

上一講 我們講解了服務發現的相關邏輯，所謂服務發現 其實就是登錄檔抓取，服務例項預設每隔30s去註冊中心抓取一下注冊表增量資料，然後合併本地登錄檔資料，最後有個hash對比的操作。

本講目錄

今天主要是看下服務續約的邏輯，服務續約就是client端給server端傳送心跳檢測，告訴對方我還活著。現在很多分散式系統都會有心跳檢查的機制，這裡一起來學習下Eureka是怎麼做心跳檢查的。

目錄如下：

1. client端心跳檢查排程任務
2. server端接收心跳檢查，設定最後renew時間

這一講內容不太多，因為上一篇文章寫全量和增量登錄檔資訊內容有點多，所以這裡將部落格儘量一篇保持一個知識點，後面還會講服務例項下線、摘除、註冊中心自我保護等機制的實現原理。

說明

原創不易，如若轉載 請標明來源：一枝花算不算浪漫

原始碼分析

client端心跳檢查排程任務

服務例項續約程式碼比較簡單，這裡還是從DiscovertClient.java 開始，很多原始碼的入口都是在這裡，因為client端初始化、註冊 都是走的這裡，因為前幾篇文章對這個類已經分析很多了，這裡只擷取部分重要程式碼：

DiscovertClient.java 初始化後 會繼續初始化一些排程任務：

private void initScheduledTasks() {
    if (clientConfig.shouldRegisterWithEureka()) {
        //  預設也是30s
        int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
        int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
        logger.info("Starting heartbeat executor: " + "renew interval is: " + renewalIntervalInSecs);

        // Heartbeat timer
        // 執行heartbeatExecutor心跳檢查，預設是30s
        scheduler.schedule(
                new TimedSupervisorTask(
                        "heartbeat",
                        scheduler,
                        heartbeatExecutor,
                        renewalIntervalInSecs,
                        TimeUnit.SECONDS,
                        expBackOffBound,
                        new HeartbeatThread()
                ),
                renewalIntervalInSecs, TimeUnit.SECONDS);

        // 執行執行緒
        instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
    } 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 {
        httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
        logger.debug("{} - Heartbeat status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
        if (httpResponse.getStatusCode() == 404) {
            REREGISTER_COUNTER.increment();
            logger.info("{} - Re-registering apps/{}", PREFIX + appPathIdentifier, instanceInfo.getAppName());
            long timestamp = instanceInfo.setIsDirtyWithTime();
            boolean success = register();
            if (success) {
                instanceInfo.unsetIsDirty(timestamp);
            }
            return success;
        }
        return httpResponse.getStatusCode() == 200;
    } catch (Throwable e) {
        logger.error("{} - was unable to send heartbeat!", PREFIX + appPathIdentifier, e);
        return false;
    }
}

public EurekaHttpResponse<InstanceInfo> sendHeartBeat(String appName, String id, InstanceInfo info, InstanceStatus overriddenStatus) {
    String urlPath = "apps/" + appName + '/' + id;
    Response response = null;
    try {
        WebTarget webResource = jerseyClient.target(serviceUrl)
                .path(urlPath)
                .queryParam("status", info.getStatus().toString())
                .queryParam("lastDirtyTimestamp", info.getLastDirtyTimestamp().toString());
        if (overriddenStatus != null) {
            webResource = webResource.queryParam("overriddenstatus", overriddenStatus.name());
        }
        Builder requestBuilder = webResource.request();
        addExtraProperties(requestBuilder);
        addExtraHeaders(requestBuilder);
        requestBuilder.accept(MediaType.APPLICATION_JSON_TYPE);
        response = requestBuilder.put(Entity.entity("{}", MediaType.APPLICATION_JSON_TYPE)); // Jersey2 refuses to handle PUT with no body
        EurekaHttpResponseBuilder<InstanceInfo> eurekaResponseBuilder = anEurekaHttpResponse(response.getStatus(), InstanceInfo.class).headers(headersOf(response));
        if (response.hasEntity()) {
            eurekaResponseBuilder.entity(response.readEntity(InstanceInfo.class));
        }
        return eurekaResponseBuilder.build();
    } finally {
        if (logger.isDebugEnabled()) {
            logger.debug("Jersey2 HTTP PUT {}/{}; statusCode={}", serviceUrl, urlPath, response == null ? "N/A" : response.getStatus());
        }
        if (response != null) {
            response.close();
        }
    }
}
 

這裡的流程很簡單，初始化DiscoveryClient 後會新建一個排程任務，然後執行HeartbeatThread中的run方法，預設是renewalIntervalInSecs 30s執行一次。
具體就是給Server端傳送一個http請求，類似於：http://localhost:8080/v2/apps/ServiceA/i-000000-1, 走的是put請求。
最後拿到響應結果，續約成功後會更新lastSuccessfulHeartbeatTimestamp 最近成功心跳檢測的時間戳。

server端接收心跳檢查請求

前幾篇文章已經說過，Server端接收http請求的入口在eureka-core模組下的 resource

包裡面，這裡直接找到ApplicationResource.java中的getInstanceInfo 方法，這裡直接請求的InstanceResource 類的構造方法，找到這個方法中的@PUT請求。可以直接看下程式碼：

InstanceResource.renewLease +AbstractInstanceRegistry.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);
    boolean isSuccess = registry.renew(app.getName(), id, isFromReplicaNode);

    // 省略部分程式碼

    logger.debug("Found (Renew): {} - {}; reply status={}" + app.getName(), id, response.getStatus());
    return response;
}


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) {
        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)) {
                Object[] args = {
                        instanceInfo.getStatus().name(),
                        instanceInfo.getOverriddenStatus().name(),
                        instanceInfo.getId()
                };
                logger.info(
                        "The instance status {} is different from overridden instance status {} for instance {}. "
                                + "Hence setting the status to overridden status", args);
                instanceInfo.setStatusWithoutDirty(overriddenInstanceStatus);
            }
        }
        renewsLastMin.increment();
        leaseToRenew.renew();
        return true;
    }
}

這裡主要看renew方法， 這裡看到registry 是一個登錄檔，通過appName獲取對應的服務登錄檔資訊。

這裡主要還是看leaseToRenew.renew() 其實很簡單，就是設定當前示例登錄檔的renew屬性的lastUpdateTimestamp 為最新時間+duration。

至於這裡的duration 我們下一講會詳細講解，duration 和服務例項摘除有關。

總結

（1）DiscoveryClient初始化的時候，會去排程一堆定時任務，其中有一個就是HeartbeatThread，心跳執行緒

（2）在這裡可以看到，預設是每隔30秒去傳送一次心跳，每隔30秒執行一次HeartbeatTHread執行緒的邏輯，傳送心跳

（3）這邊的話就是去傳送這個心跳，走的是EurekaHttpClient的sendHeartbeat()方法，http://localhost:8080/v2/apps/ServiceA/i-000000-1，走的是put請求

（4）負責承接服務例項的心跳相關的這些操作的，是ApplicationsResource，服務相關的controller。找到ApplicationResource，再次找到InstanceResource，通過PUT請求，可以找到renewLease方法。

（5）通過登錄檔的renew()方法，進去完成服務續約，實際進入AbstractInstanceRegistry的renew()方法

（6）從登錄檔的map中，根據服務名和例項id，獲取一個Lease，實際的服務續約的邏輯，其實就是在Lease物件中，更新一下lastUpdateTimestamp這個時間戳，每次續約，就更新一下這個時間戳就ok了。

申明

本文章首發自本人部落格：https://www.cnblogs.com/wang-meng 和公眾號：壹枝花算不算浪漫，如若轉載請標明來源!

感興趣的小夥伴可關注個人公眾號：壹枝花算不算浪漫

相關推薦