Eureka源碼分析(十) 全量獲取

這次我們說(shuō)一下eureka的全量獲取。EurekaClient 啟動(dòng)時(shí),首先執(zhí)行一次全量獲取進(jìn)行本地緩存注冊(cè)信息。

localRegionApps.set(new Applications());
if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) {
        fetchRegistryFromBackup();
    }

配置 eureka.shouldFetchRegistry = true,開(kāi)啟從 EurekaServer獲取注冊(cè)信息泳梆。默認(rèn)值:true。
調(diào)用 fetchRegistry(false)方法榜掌,從 Eureka-Server 全量獲取注冊(cè)信息优妙。
EurekaClient在初始化過(guò)程中,創(chuàng)建獲取注冊(cè)信息線程憎账,固定間隔向 Eureka-Server 發(fā)起獲取注冊(cè)信息( fetch )套硼,刷新本地注冊(cè)信息緩存

private void initScheduledTasks() {
    // 從 Eureka-Server 拉取注冊(cè)信息執(zhí)行器
    if (clientConfig.shouldFetchRegistry()) {
        // registry cache refresh timer
        int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
        int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
        scheduler.schedule(
                new TimedSupervisorTask(
                        "cacheRefresh",
                        scheduler,
                        cacheRefreshExecutor,
                        registryFetchIntervalSeconds,
                        TimeUnit.SECONDS,
                        expBackOffBound,
                        new CacheRefreshThread()
                ),
                registryFetchIntervalSeconds, TimeUnit.SECONDS);
    }

    // 向 Eureka-Server 心跳(續(xù)租)執(zhí)行器
    if (clientConfig.shouldRegisterWithEureka()) {
        int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs(); // 續(xù)租頻率
        int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound(); // 系數(shù)
        logger.info("Starting heartbeat executor: " + "renew interval is: " + renewalIntervalInSecs);

        // Heartbeat timer
        scheduler.schedule(
                new TimedSupervisorTask(
                        "heartbeat",
                        scheduler,
                        heartbeatExecutor,
                        renewalIntervalInSecs,
                        TimeUnit.SECONDS,
                        expBackOffBound,
                        new HeartbeatThread()
                ),
                renewalIntervalInSecs, TimeUnit.SECONDS);

        // 創(chuàng)建 應(yīng)用實(shí)例信息復(fù)制器
        // InstanceInfo replicator
        instanceInfoReplicator = new InstanceInfoReplicator(
                this,
                instanceInfo,
                clientConfig.getInstanceInfoReplicationIntervalSeconds(),
                2); // burstSize

        // 創(chuàng)建 應(yīng)用實(shí)例狀態(tài)變更監(jiān)聽(tīng)器
        statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
            @Override
            public String getId() {
                return "statusChangeListener";
            }

            @Override
            public void notify(StatusChangeEvent statusChangeEvent) {
                if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||
                        InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) {
                    // log at warn level if DOWN was involved
                    logger.warn("Saw local status change event {}", statusChangeEvent);
                } else {
                    logger.info("Saw local status change event {}", statusChangeEvent);
                }
                instanceInfoReplicator.onDemandUpdate();
            }
        };

        // 注冊(cè) 應(yīng)用實(shí)例狀態(tài)變更監(jiān)聽(tīng)器
        if (clientConfig.shouldOnDemandUpdateStatusChange()) {
            applicationInfoManager.registerStatusChangeListener(statusChangeListener);
        }

        // 開(kāi)啟 應(yīng)用實(shí)例信息復(fù)制器
        instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
    } else {
        logger.info("Not registering with Eureka server per configuration");
    }
}

調(diào)用 refreshRegistry(false)方法,刷新注冊(cè)信息緩存

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) {
            // 設(shè)置 注冊(cè)信息的應(yīng)用實(shí)例數(shù)
            registrySize = localRegionApps.get().size();
            // 設(shè)置 最后獲取注冊(cè)信息時(shí)間
            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.toString());
        }
    } catch (Throwable e) {
        logger.error("Cannot fetch registry from server", e);
    }        
}

調(diào)用 fetchRegistry方法胞皱,從EurekaServer獲取注冊(cè)信息

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

    try {
        // 獲取 本地緩存的注冊(cè)的應(yīng)用實(shí)例集合
        // If the delta is disabled or if it is the first time, get all
        // applications
        Applications applications = getApplications();

        // 全量獲取
        if (clientConfig.shouldDisableDelta() // 禁用增量獲取
                || (!Strings.isNullOrEmpty(clientConfig.getRegistryRefreshSingleVipAddress()))
                || forceFullRegistryFetch
                || (applications == null) // 空
                || (applications.getRegisteredApplications().size() == 0) // 空
                || (applications.getVersion() == -1)) //Client application does not have latest library supporting delta
        {
            logger.info("Disable delta property : {}", clientConfig.shouldDisableDelta());
            logger.info("Single vip registry refresh property : {}", clientConfig.getRegistryRefreshSingleVipAddress());
            logger.info("Force full registry fetch : {}", forceFullRegistryFetch);
            logger.info("Application is null : {}", (applications == null));
            logger.info("Registered Applications size is zero : {}",
                    (applications.getRegisteredApplications().size() == 0));
            logger.info("Application version is -1: {}", (applications.getVersion() == -1));
            // 執(zhí)行 全量獲取
            getAndStoreFullRegistry();
        } else {
            // 執(zhí)行 增量獲取
            getAndUpdateDelta(applications);
        }
        // 設(shè)置 應(yīng)用集合 hashcode
        applications.setAppsHashCode(applications.getReconcileHashCode());
        // 打印 本地緩存的注冊(cè)的應(yīng)用實(shí)例數(shù)量
        logTotalInstances();
    } catch (Throwable e) {
        logger.error(PREFIX + appPathIdentifier + " - was unable to refresh its cache! status = " + e.getMessage(), e);
        return false;
    } finally {
        if (tracer != null) {
            tracer.stop();
        }
    }

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

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

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

調(diào)用 getAndStoreFullRegistry方法邪意,全量獲取注冊(cè)信息,并設(shè)置到本地緩存

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

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

    // 全量獲取注冊(cè)信息
    Applications apps = null;
    EurekaHttpResponse<Applications> httpResponse = clientConfig.getRegistryRefreshSingleVipAddress() == null
            ? 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());

    // 設(shè)置到本地緩存
    if (apps == null) {
        logger.error("The application is null for some reason. Not storing this information");
    } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
        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");
    }
}

ApplicationsResource反砌,處理所有應(yīng)用的請(qǐng)求操作的 Resource ( Controller )抄罕。

接收全量獲取請(qǐng)求,映射 getContainers方法

@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) {
    // TODO[0009]:RemoteRegionRegistry
    boolean isRemoteRegionRequested = null != regionsStr && !regionsStr.isEmpty();
    String[] regions = null;
    if (!isRemoteRegionRequested) {
        EurekaMonitors.GET_ALL.increment();
    } else {
        regions = regionsStr.toLowerCase().split(",");
        Arrays.sort(regions); // So we don't have different caches for same regions queried in different order.
        EurekaMonitors.GET_ALL_WITH_REMOTE_REGIONS.increment();
    }

    // 判斷是否可以訪問(wèn)
    // Check if the server allows the access to the registry. The server can
    // restrict access if it is not
    // ready to serve traffic depending on various reasons.
    if (!registry.shouldAllowAccess(isRemoteRegionRequested)) {
        return Response.status(Status.FORBIDDEN).build();
    }

    // API 版本
    CurrentRequestVersion.set(Version.toEnum(version));

    // 返回?cái)?shù)據(jù)格式
    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;
    }

    // 響應(yīng)緩存鍵( KEY )
    Key cacheKey = new Key(Key.EntityType.Application,
            ResponseCacheImpl.ALL_APPS,
            keyType, CurrentRequestVersion.get(), EurekaAccept.fromString(eurekaAccept), regions
    );

    // 響應(yīng)緩存結(jié)果
    Response response;
    if (acceptEncoding != null && acceptEncoding.contains(HEADER_GZIP_VALUE)) {
        response = Response.ok(responseCache.getGZIP(cacheKey))
                .header(HEADER_CONTENT_ENCODING, HEADER_GZIP_VALUE)
                .header(HEADER_CONTENT_TYPE, returnMediaType)
                .build();
    } else {
        response = Response.ok(responseCache.get(cacheKey))
                .build();
    }
    return response;
}

ResponseCacheImpl于颖,響應(yīng)緩存實(shí)現(xiàn)類。

在 ResponseCacheImpl 里嚷兔,將緩存拆分成兩層 :

只讀緩存( readOnlyCacheMap )
固定過(guò)期 + 固定大小的讀寫緩存( readWriteCacheMap )

緩存過(guò)期策略如下:

應(yīng)用實(shí)例注冊(cè)森渐、下線做入、過(guò)期時(shí),只過(guò)期 readWriteCacheMap 同衣。
readWriteCacheMap 寫入一段時(shí)間( 可配置 )后自動(dòng)過(guò)期竟块。
定時(shí)任務(wù)對(duì)比 readWriteCacheMap 和 readOnlyCacheMap 的緩存值,若不一致耐齐,以前者為主浪秘。通過(guò)這樣的方式,實(shí)現(xiàn)了 readOnlyCacheMap 的定時(shí)過(guò)期埠况。
調(diào)用ResponseCacheImpl的get方法耸携,獲取緩存

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

調(diào)用 AbstractInstanceRegistry的getApplications方法,獲取注冊(cè)的應(yīng)用集合

public Applications getApplications() {
    boolean disableTransparentFallback = serverConfig.disableTransparentFallbackToOtherRegion();
    if (disableTransparentFallback) { // TODO[0009]:RemoteRegionRegistry
        return getApplicationsFromLocalRegionOnly();
    } else {
        return getApplicationsFromAllRemoteRegions();  // Behavior of falling back to remote region can be disabled.
    }
}

定時(shí)任務(wù)對(duì)比 readWriteCacheMap 和 readOnlyCacheMap 的緩存值辕翰,若不一致夺衍,以前者為主。通過(guò)這樣的方式喜命,實(shí)現(xiàn)了 readOnlyCacheMap 的定時(shí)過(guò)期沟沙。

private TimerTask getCacheUpdateTask() {
    return new TimerTask() {
        @Override
        public void run() {
            logger.debug("Updating the client cache from response cache");
            for (Key key : readOnlyCacheMap.keySet()) { // 循環(huán) readOnlyCacheMap 的緩存鍵
                if (logger.isDebugEnabled()) {
                    Object[] args = {key.getEntityType(), key.getName(), key.getVersion(), key.getType()};
                    logger.debug("Updating the client cache from response cache for key : {} {} {} {}", args);
                }
                try {
                    CurrentRequestVersion.set(key.getVersion());
                    Value cacheValue = readWriteCacheMap.get(key);
                    Value currentCacheValue = readOnlyCacheMap.get(key);
                    if (cacheValue != currentCacheValue) { // 不一致時(shí),進(jìn)行替換
                        readOnlyCacheMap.put(key, cacheValue);
                    }
                } catch (Throwable th) {
                    logger.error("Error while updating the client cache from response cache for key {}", key.toStringCompact(), th);
                }
            }
        }
    };
}

eureka的全量獲取就完成了壁榕。

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
  • 序言:七十年代末矛紫,一起剝皮案震驚了整個(gè)濱河市,隨后出現(xiàn)的幾起案子牌里,更是在濱河造成了極大的恐慌颊咬,老刑警劉巖,帶你破解...
    沈念sama閱讀 211,639評(píng)論 6 492
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件二庵,死亡現(xiàn)場(chǎng)離奇詭異贪染,居然都是意外死亡,警方通過(guò)查閱死者的電腦和手機(jī)催享,發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 90,277評(píng)論 3 385
  • 文/潘曉璐 我一進(jìn)店門杭隙,熙熙樓的掌柜王于貴愁眉苦臉地迎上來(lái),“玉大人因妙,你說(shuō)我怎么就攤上這事痰憎。” “怎么了攀涵?”我有些...
    開(kāi)封第一講書人閱讀 157,221評(píng)論 0 348
  • 文/不壞的土叔 我叫張陵铣耘,是天一觀的道長(zhǎng)。 經(jīng)常有香客問(wèn)我以故,道長(zhǎng)蜗细,這世上最難降的妖魔是什么? 我笑而不...
    開(kāi)封第一講書人閱讀 56,474評(píng)論 1 283
  • 正文 為了忘掉前任,我火速辦了婚禮炉媒,結(jié)果婚禮上踪区,老公的妹妹穿的比我還像新娘。我一直安慰自己吊骤,他們只是感情好缎岗,可當(dāng)我...
    茶點(diǎn)故事閱讀 65,570評(píng)論 6 386
  • 文/花漫 我一把揭開(kāi)白布。 她就那樣靜靜地躺著白粉,像睡著了一般传泊。 火紅的嫁衣襯著肌膚如雪。 梳的紋絲不亂的頭發(fā)上鸭巴,一...
    開(kāi)封第一講書人閱讀 49,816評(píng)論 1 290
  • 那天眷细,我揣著相機(jī)與錄音,去河邊找鬼奕扣。 笑死薪鹦,一個(gè)胖子當(dāng)著我的面吹牛,可吹牛的內(nèi)容都是我干的惯豆。 我是一名探鬼主播池磁,決...
    沈念sama閱讀 38,957評(píng)論 3 408
  • 文/蒼蘭香墨 我猛地睜開(kāi)眼,長(zhǎng)吁一口氣:“原來(lái)是場(chǎng)噩夢(mèng)啊……” “哼楷兽!你這毒婦竟也來(lái)了地熄?” 一聲冷哼從身側(cè)響起,我...
    開(kāi)封第一講書人閱讀 37,718評(píng)論 0 266
  • 序言:老撾萬(wàn)榮一對(duì)情侶失蹤芯杀,失蹤者是張志新(化名)和其女友劉穎端考,沒(méi)想到半個(gè)月后,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體揭厚,經(jīng)...
    沈念sama閱讀 44,176評(píng)論 1 303
  • 正文 獨(dú)居荒郊野嶺守林人離奇死亡却特,尸身上長(zhǎng)有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 36,511評(píng)論 2 327
  • 正文 我和宋清朗相戀三年,在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了筛圆。 大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片裂明。...
    茶點(diǎn)故事閱讀 38,646評(píng)論 1 340
  • 序言:一個(gè)原本活蹦亂跳的男人離奇死亡,死狀恐怖太援,靈堂內(nèi)的尸體忽然破棺而出闽晦,到底是詐尸還是另有隱情,我是刑警寧澤提岔,帶...
    沈念sama閱讀 34,322評(píng)論 4 330
  • 正文 年R本政府宣布仙蛉,位于F島的核電站,受9級(jí)特大地震影響碱蒙,放射性物質(zhì)發(fā)生泄漏荠瘪。R本人自食惡果不足惜,卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 39,934評(píng)論 3 313
  • 文/蒙蒙 一、第九天 我趴在偏房一處隱蔽的房頂上張望哀墓。 院中可真熱鬧鞭莽,春花似錦、人聲如沸麸祷。這莊子的主人今日做“春日...
    開(kāi)封第一講書人閱讀 30,755評(píng)論 0 21
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽(yáng)阶牍。三九已至,卻和暖如春星瘾,著一層夾襖步出監(jiān)牢的瞬間走孽,已是汗流浹背。 一陣腳步聲響...
    開(kāi)封第一講書人閱讀 31,987評(píng)論 1 266
  • 我被黑心中介騙來(lái)泰國(guó)打工琳状, 沒(méi)想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留磕瓷,地道東北人。 一個(gè)月前我還...
    沈念sama閱讀 46,358評(píng)論 2 360
  • 正文 我出身青樓念逞,卻偏偏與公主長(zhǎng)得像困食,于是被迫代替她去往敵國(guó)和親。 傳聞我的和親對(duì)象是個(gè)殘疾皇子翎承,可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 43,514評(píng)論 2 348

推薦閱讀更多精彩內(nèi)容