Spring Cloud Eureka的服務與列表獲取的方法是什么

本篇內容主要講解“Spring Cloud Eureka的服務與列表獲取的方法是什么”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“Spring Cloud Eureka的服務與列表獲取的方法是什么”吧!

關于服務與實例列表獲取

EurekaClient端

我們從Ribbon說起：EurekaClient也存在緩存，應用服務實例列表信息在每個EurekaClient服務消費端都有緩存。一般的，Ribbon的LoadBalancer會讀取這個緩存，來知道當前有哪些實例可以調用，從而進行負載均衡。這個loadbalancer同樣也有緩存。

首先看這個LoadBalancer的緩存更新機制，相關類是PollingServerListUpdater：

final Runnable wrapperRunnable = new Runnable() {
    @Override
    public void run() {
        if (!isActive.get()) {
            if (scheduledFuture != null) {
                scheduledFuture.cancel(true);
            }
            return;
        }
        try {
            //從EurekaClient緩存中獲取服務實例列表，保存在本地緩存
            updateAction.doUpdate();
            lastUpdated = System.currentTimeMillis();
        } catch (Exception e) {
            logger.warn("Failed one update cycle", e);
        }
    }
};

//定時調度
scheduledFuture = getRefreshExecutor().scheduleWithFixedDelay(
        wrapperRunnable,
        initialDelayMs,
        refreshIntervalMs,
        TimeUnit.MILLISECONDS
);

這個updateAction.doUpdate();就是從EurekaClient緩存中獲取服務實例列表，保存在BaseLoadBalancer的本地緩存：

protected volatile List<Server> allServerList = Collections.synchronizedList(new ArrayList<Server>());

public void setServersList(List lsrv) {
    //寫入allServerList的代碼，這里略
}

@Override
public List<Server> getAllServers() {
    return Collections.unmodifiableList(allServerList);
}

這里的getAllServers會在每個負載均衡規(guī)則中被調用，例如RoundRobinRule：

public Server choose(ILoadBalancer lb, Object key) {
    if (lb == null) {
        log.warn("no load balancer");
        return null;
    }

    Server server = null;
    int count = 0;
    while (server == null && count++ < 10) {
        List<Server> reachableServers = lb.getReachableServers();
        //獲取服務實例列表，調用的就是剛剛提到的getAllServers
        List<Server> allServers = lb.getAllServers();
        int upCount = reachableServers.size();
        int serverCount = allServers.size();

        if ((upCount == 0) || (serverCount == 0)) {
            log.warn("No up servers available from load balancer: " + lb);
            return null;
        }

        int nextServerIndex = incrementAndGetModulo(serverCount);
        server = allServers.get(nextServerIndex);

        if (server == null) {
            /* Transient. */
            Thread.yield();
            continue;
        }

        if (server.isAlive() && (server.isReadyToServe())) {
            return (server);
        }

        // Next.
        server = null;
    }

    if (count >= 10) {
        log.warn("No available alive servers after 10 tries from load balancer: "
                + lb);
    }
    return server;
}

這個緩存需要注意下，有時候我們只修改了EurekaClient緩存的更新時間，但是沒有修改這個LoadBalancer的刷新本地緩存時間，就是ribbon.ServerListRefreshInterval,這個參數(shù)可以設置的很小，因為沒有從網絡讀取，就是從一個本地緩存刷到另一個本地緩存（如何配置緩存配置來實現(xiàn)服務實例快速下線快速感知快速刷新，可以參考我的另一篇文章）。

然后我們來看一下EurekaClient本身的緩存，直接看關鍵類DiscoveryClient的相關源碼，我們這里只關心本地Region的，多Region配置我們先忽略：

//本地緩存，可以理解為是一個軟鏈接
private final AtomicReference<Applications> localRegionApps = new AtomicReference<Applications>();

private void initScheduledTasks() {
    //如果配置為需要拉取服務列表，則設置定時拉取任務，這個配置默認是需要拉取服務列表
    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);
    }
    //其他定時任務初始化的代碼，忽略
}

//定時從EurekaServer拉取服務列表的任務
class CacheRefreshThread implements Runnable {
        public void run() {
            refreshRegistry();
        }
}

void refreshRegistry() {
    try {
        //多Region配置處理代碼，忽略

        boolean success = fetchRegistry(remoteRegionsModified);
        if (success) {
            registrySize = localRegionApps.get().size();
            lastSuccessfulRegistryFetchTimestamp = System.currentTimeMillis();
        }

        //日志代碼，忽略
    } catch (Throwable e) {
        logger.error("Cannot fetch registry from server", e);
    }        
}

//定時從EurekaServer拉取服務列表的核心方法
private boolean fetchRegistry(boolean forceFullRegistryFetch) {
    Stopwatch tracer = FETCH_REGISTRY_TIMER.start();

    try {
        Applications applications = getApplications();

        //判斷，如果是第一次拉取，或者app列表為空，就進行全量拉取，否則就會進行增量拉取
        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.error(PREFIX + appPathIdentifier + " - was unable to refresh its cache! status = " + e.getMessage(), e);
        return false;
    } finally {
        if (tracer != null) {
            tracer.stop();
        }
    }

    //緩存更新完成，發(fā)送個event給觀察者，目前沒啥用 
    onCacheRefreshed();

    // 檢查下遠端的服務實例列表里面包括自己，并且狀態(tài)是否對，這里我們不關心
    updateInstanceRemoteStatus();

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

//全量拉取代碼
private void getAndStoreFullRegistry() throws Throwable {
    long currentUpdateGeneration = fetchRegistryGeneration.get();

    Applications apps = null;
    //訪問/eureka/apps接口，拉取所有服務實例信息
    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());

    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");
    }
}

//增量拉取代碼

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

    Applications delta = null;
    //訪問/eureka/delta接口，拉取所有服務實例增量信息
    EurekaHttpResponse<Applications> httpResponse = eurekaTransport.queryClient.getDelta(remoteRegionsRef.get());
    if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
        delta = httpResponse.getEntity();
    }

    if (delta == null) {
        //如果delta為空，拉取增量失敗，就全量拉取
        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)) {
        //這里設置原子鎖的原因是怕某次調度網絡請求時間過長，導致同一時間有多線程拉取到增量信息并發(fā)修改
        //拉取增量成功，檢查hashcode是否一樣，不一樣的話也會全量拉取
        logger.debug("Got delta update with apps hashcode {}", delta.getAppsHashCode());
        String reconcileHashCode = "";
        if (fetchRegistryUpdateLock.tryLock()) {
            try {
                updateDelta(delta);
                reconcileHashCode = getReconcileHashCode(applications);
            } finally {
                fetchRegistryUpdateLock.unlock();
            }
        } else {
            logger.warn("Cannot acquire update lock, aborting getAndUpdateDelta");
        }
        // There is a diff in number of instances for some reason
        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());
    }
}

以上就是對于EurekaClient拉取服務實例信息的源代碼分析，總結EurekaClient 重要緩存如下：

1. EurekaClient第一次全量拉取，定時增量拉取應用服務實例信息，保存在緩存中。

2. EurekaClient增量拉取失敗，或者增量拉取之后對比hashcode發(fā)現(xiàn)不一致，就會執(zhí)行全量拉取，這樣避免了網絡某時段分片帶來的問題。

3. 同時對于服務調用，如果涉及到ribbon負載均衡，那么ribbon對于這個實例列表也有自己的緩存，這個緩存定時從EurekaClient的緩存更新

EurekaServer端

在EurekaServer端，所有的讀取請求都是讀的ReadOnlyMap（這個可以配置） 有定時任務會定時從ReadWriteMap同步到ReadOnlyMap這個時間配置是：

#eureka server刷新readCacheMap的時間，注意，client讀取的是readCacheMap，這個時間決定了多久會把readWriteCacheMap的緩存更新到readCacheMap上
#默認30s
eureka.server.responseCacheUpdateInvervalMs=3000

相關代碼：

if (shouldUseReadOnlyResponseCache) {
            timer.schedule(getCacheUpdateTask(),
                    new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
                            + responseCacheUpdateIntervalMs),
                    responseCacheUpdateIntervalMs);
        }
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()) {
                    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) {
                        readOnlyCacheMap.put(key, cacheValue);
                    }
                } catch (Throwable th) {
                    logger.error("Error while updating the client cache from response cache", th);
                }
            }
        }
    };
}

ReadWriteMap是一個LoadingCache，將Registry中的服務實例信息封裝成要返回的http響應（分別是經過gzip壓縮和非壓縮的），同時還有兩個特殊key，ALL_APPS和ALL_APPS_DELTA ALL_APPS就是所有服務實例信息 ALL_APPS_DELTA就是之前講注冊說的RecentlyChangedQueue里面的實例列表封裝的http響應信息

到此，相信大家對“Spring Cloud Eureka的服務與列表獲取的方法是什么”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續(xù)學習！