【一起學原始碼-微服務】Nexflix Eureka 原始碼七：通過單元測試來Debug Eureka註冊過程

前言

上一講eureka client是如何註冊的，一直跟到原始碼傳送http請求為止，當時看eureka client註冊時如此費盡，光是找一個regiter的地方就找了半天，那麼client端傳送了http請求給server端，server端是如何處理的呢？

帶著這麼一個疑問就開始今天原始碼的解讀了。

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

原始碼解讀

從何讀起？

上一講我們知道，跟進client註冊一直到 AbstractJersey2EurekaHttpClient.register方法，這裡先看下其中的原始碼：

public EurekaHttpResponse<Void> register(InstanceInfo info) {
        String urlPath = "apps/" + info.getAppName();
        Response response = null;
        try {
            // 傳送請求，類似於：http://localhost:8080/v2/apps/ServiceA
            // 傳送的是post請求，服務例項的物件被打成了一個json傳送，包括自己的主機、ip、埠號
            // eureka server 就知道了這個ServiceA這個服務，有一個服務例項，比如是在192.168.31.109、host-01、8761埠
            Builder resourceBuilder = jerseyClient.target(serviceUrl).path(urlPath).request();
            addExtraProperties(resourceBuilder);
            addExtraHeaders(resourceBuilder);
            response = resourceBuilder
                    .accept(MediaType.APPLICATION_JSON)
                    .acceptEncoding("gzip")
                    .post(Entity.json(info));
            return anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
        } finally {
            if (logger.isDebugEnabled()) {
                logger.debug("Jersey2 HTTP POST {}/{} with instance {}; statusCode={}", serviceUrl, urlPath, info.getId(),
                        response == null ? "N/A" : response.getStatus());
            }
            if (response != null) {
                response.close();
            }
        }
    }

那這種情況我們肯定可以猜測，server端應該有個controller來接收此http請求，然後默默的去做一些註冊的邏輯。

緊接著我們從/apps/這個關鍵詞入手，進行全域性搜尋：

全域性搜尋結果如下，這裡可以看到很多test 呼叫，這裡框起來的一個是不是類似於我們controller介面的呼叫呢？直接點進去檢視，然後一步步跟進。

原始碼分析

接著上面說的，跟進ApplicationResource這個類，可以找到如下方法：

@Path("{appId}")
public ApplicationResource getApplicationResource(
        @PathParam("version") String version,
        @PathParam("appId") String appId) {
    CurrentRequestVersion.set(Version.toEnum(version));
    return new ApplicationResource(appId, serverConfig, registry);
}

這個appId可以理解為我們之前傳遞的appName，緊接著這裡是直接構造了一個ApplicationResource例項，接著跟進程式碼，進入ApplicationResource中我們可以看到很多@GET、@POST 等restful介面，還記得上面我們register方法中，傳送的http請求用的就是POST方法，所以我們這裡直接看@POST請求

@POST
@Consumes({"application/json", "application/xml"})
public Response addInstance(InstanceInfo info,
                            @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) {
    logger.debug("Registering instance {} (replication={})", info.getId(), isReplication);
    // validate that the instanceinfo contains all the necessary required fields
    if (isBlank(info.getId())) {
        return Response.status(400).entity("Missing instanceId").build();
    } else if (isBlank(info.getHostName())) {
        return Response.status(400).entity("Missing hostname").build();
    } else if (isBlank(info.getIPAddr())) {
        return Response.status(400).entity("Missing ip address").build();
    } else if (isBlank(info.getAppName())) {
        return Response.status(400).entity("Missing appName").build();
    } else if (!appName.equals(info.getAppName())) {
        return Response.status(400).entity("Mismatched appName, expecting " + appName + " but was " + info.getAppName()).build();
    } else if (info.getDataCenterInfo() == null) {
        return Response.status(400).entity("Missing dataCenterInfo").build();
    } else if (info.getDataCenterInfo().getName() == null) {
        return Response.status(400).entity("Missing dataCenterInfo Name").build();
    }

    // handle cases where clients may be registering with bad DataCenterInfo with missing data
    DataCenterInfo dataCenterInfo = info.getDataCenterInfo();
    if (dataCenterInfo instanceof UniqueIdentifier) {
        String dataCenterInfoId = ((UniqueIdentifier) dataCenterInfo).getId();
        if (isBlank(dataCenterInfoId)) {
            boolean experimental = "true".equalsIgnoreCase(serverConfig.getExperimental("registration.validation.dataCenterInfoId"));
            if (experimental) {
                String entity = "DataCenterInfo of type " + dataCenterInfo.getClass() + " must contain a valid id";
                return Response.status(400).entity(entity).build();
            } else if (dataCenterInfo instanceof AmazonInfo) {
                AmazonInfo amazonInfo = (AmazonInfo) dataCenterInfo;
                String effectiveId = amazonInfo.get(AmazonInfo.MetaDataKey.instanceId);
                if (effectiveId == null) {
                    amazonInfo.getMetadata().put(AmazonInfo.MetaDataKey.instanceId.getName(), info.getId());
                }
            } else {
                logger.warn("Registering DataCenterInfo of type {} without an appropriate id", dataCenterInfo.getClass());
            }
        }
    }

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

由於程式碼不是很長，這裡都給截取出來了。其實這裡做的事情就很簡單了。

做一些常規的chek，檢查註冊例項InstanceInfo的一些基本資訊
DataCenter的相關操作，這裡還涉及到亞馬遜雲，我們直接跳過
registry.register(info, "true".equals(isReplication)); 這裡才是核心的註冊，我們繼續往下

public void register(final InstanceInfo info, final boolean isReplication) {
    int leaseDuration = Lease.DEFAULT_DURATION_IN_SECS;
    if (info.getLeaseInfo() != null && info.getLeaseInfo().getDurationInSecs() > 0) {
        leaseDuration = info.getLeaseInfo().getDurationInSecs();
    }
    super.register(info, leaseDuration, isReplication);
    replicateToPeers(Action.Register, info.getAppName(), info.getId(), info, null, isReplication);
}

public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
    try {
        read.lock();
        Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
        REGISTER.increment(isReplication);
        if (gMap == null) {
            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);

            // this is a > instead of a >= because if the timestamps are equal, we still take the remote transmitted
            // InstanceInfo instead of the server local copy.
            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 {
            // The lease does not exist and hence it is a new registration
            synchronized (lock) {
                if (this.expectedNumberOfRenewsPerMin > 0) {
                    // Since the client wants to cancel it, reduce the threshold
                    // (1
                    // for 30 seconds, 2 for a minute)
                    this.expectedNumberOfRenewsPerMin = this.expectedNumberOfRenewsPerMin + 2;
                    this.numberOfRenewsPerMinThreshold =
                            (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());
                }
            }
            logger.debug("No previous lease information found; it is new registration");
        }
        Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
        if (existingLease != null) {
            lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
        }
        gMap.put(registrant.getId(), lease);
        synchronized (recentRegisteredQueue) {
            recentRegisteredQueue.add(new Pair<Long, String>(
                    System.currentTimeMillis(),
                    registrant.getAppName() + "(" + registrant.getId() + ")"));
        }
        // This is where the initial state transfer of overridden status happens
        if (!InstanceStatus.UNKNOWN.equals(registrant.getOverriddenStatus())) {
            logger.debug("Found overridden status {} for instance {}. Checking to see if needs to be add to the "
                            + "overrides", registrant.getOverriddenStatus(), registrant.getId());
            if (!overriddenInstanceStatusMap.containsKey(registrant.getId())) {
                logger.info("Not found overridden id {} and hence adding it", registrant.getId());
                overriddenInstanceStatusMap.put(registrant.getId(), registrant.getOverriddenStatus());
            }
        }
        InstanceStatus overriddenStatusFromMap = overriddenInstanceStatusMap.get(registrant.getId());
        if (overriddenStatusFromMap != null) {
            logger.info("Storing overridden status {} from map", overriddenStatusFromMap);
            registrant.setOverriddenStatus(overriddenStatusFromMap);
        }

        // Set the status based on the overridden status rules
        InstanceStatus overriddenInstanceStatus = getOverriddenInstanceStatus(registrant, existingLease, isReplication);
        registrant.setStatusWithoutDirty(overriddenInstanceStatus);

        // If the lease is registered with UP status, set lease service up timestamp
        if (InstanceStatus.UP.equals(registrant.getStatus())) {
            lease.serviceUp();
        }
        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();
    }
}

到了這裡東西就有點多了，我們慢慢梳理。

reda.lock() 這裡使用的是讀鎖，方便多個服務例項同時來註冊
這裡關鍵資訊是registry的資料結構，同時這也是儲存註冊例項的物件。

private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry
            = new ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>();

ConcurrentHashMap的key是appName
第二層Map的key是appId，所以資料結構格式類似於：

{
    “ServiceA”: {
        “001”: Lease<InstanceInfo>,
        “002”: Lease<InstanceInfo>,
        “003”: Lease<InstanceInfo>
    },
    “ServiceB”: {
        “001”: Lease<InstanceInfo>
    }
}

這裡面還有兩個佇列recentRegisteredQueue、recentlyChangedQueue，其中registerQueue預設儲存最近1000條註冊的例項資訊。
後面就是一些狀態設定之類的操作

登錄檔使用場景

我們註冊完成之後，開啟eureka 後臺配置頁面，可以看到自己的例項已經在頁面上了，那麼這個東東是如何展示的呢？

我們都知道eureka-resources模組下有很多jsp資訊，點開status.jsp檢視一下：

這裡用到了 serverContext.getRegistry().getSortedApplications(), 然後在通過獲取的Applicaiton 去執行app.getInstances()等到了所有大的服務例項資訊。

這裡我們還需要回頭看下EurekaBootStrap中的程式碼，看看Application是如何來的。

從PeerAwareInstanceRegistryImpl.java的getSortedApplications()一直跟到 AbstractInstanceRegistry.java的getApplicationsFromMultipleRegions(),如下圖所示：

看到這裡是不是就真相大白了？
這裡再總結一下：

在jsp程式碼中，拿到了EurekaServerContext，所以之前為什麼要將這個東東放到一個Holder裡面去，就是隨時都要從這個裡面去獲取一些資料

然後會從EurekaServerContext，獲取到登錄檔，PeerAwareInstanceRegistry，登錄檔，從裡面獲取所有的服務資訊，從底層的map資料結構中，獲取所有的服務註冊的資訊，遍歷，封裝到一個叫Application的東西里去，一個Application就代表了一個服務，裡面包含很多個服務例項。

Eureka的服務註冊流程圖

申明

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

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

【一起學原始碼-微服務】Nexflix Eureka 原始碼七：通過單元測試來Debug Eureka註冊過程

前言

原始碼解讀

從何讀起？

原始碼分析

登錄檔使用場景

Eureka的服務註冊流程圖

申明

【一起學原始碼-微服務】Nexflix Eureka 原始碼七：通過單元測試來Debug Eureka註冊過程

【一起學原始碼-微服務】Nexflix Eureka 原始碼二：EurekaServer啟動之配置檔案載入以及面向介面的配置項讀取

【一起學原始碼-微服務】Nexflix Eureka 原始碼三：EurekaServer啟動之EurekaServer上下文EurekaClient建立

【一起學原始碼-微服務】Nexflix Eureka 原始碼六：在眼花繚亂的程式碼中，EurekaClient是如何註冊的？

【一起學原始碼-微服務】Nexflix Eureka 原始碼八：EurekaClient登錄檔抓取精妙設計分析！

【一起學原始碼-微服務】Nexflix Eureka 原始碼九：服務續約原始碼分析

【一起學原始碼-微服務】Nexflix Eureka 原始碼十：服務下線及例項摘除，一個client下線到底多久才會被其他例項感知？

【一起學原始碼-微服務】Nexflix Eureka 原始碼十一：EurekaServer自我保護機制竟然有這麼多Bug？

【一起學原始碼-微服務】Nexflix Eureka 原始碼十二：EurekaServer叢集模式原始碼分析

【一起學原始碼-微服務】Nexflix Eureka 原始碼十三：Eureka原始碼解讀完結撒花篇~！

【一起學原始碼-微服務】Ribbon 原始碼一：Ribbon概念理解及Demo除錯

【一起學原始碼-微服務】Ribbon 原始碼二：通過Debug找出Ribbon初始化流程及ILoadBalancer原理分析

【一起學原始碼-微服務】Ribbon 原始碼三：Ribbon與Eureka整合原理分析

【一起學原始碼-微服務】Ribbon 原始碼四：進一步探究Ribbon的IRule和IPing

【一起學原始碼-微服務】Ribbon原始碼五：Ribbon原始碼解讀彙總篇~

【一起學原始碼-微服務】Feign 原始碼一：原始碼初探，通過Demo Debug Feign原始碼

【一起學原始碼-微服務】Feign 原始碼二：Feign動態代理構造過程

【一起學原始碼-微服務】Feign 原始碼三：Feign結合Ribbon實現負載均衡的原理分析

【一起學原始碼-微服務】Hystrix 原始碼一：Hystrix基礎原理與Demo搭建

【一起學原始碼-微服務】Hystrix 原始碼二：Hystrix核心流程：Hystix非降級邏輯流程梳理

【一起學原始碼-微服務】Nexflix Eureka 原始碼七：通過單元測試來Debug Eureka註冊過程

前言

原始碼解讀

從何讀起？

原始碼分析

登錄檔使用場景

Eureka的服務註冊流程圖

申明

相關推薦