Spring Cloud進階篇之Eureka原理分析
前言
之前寫了幾篇Spring Cloud的小白教程，相信看過的朋友對Spring Cloud中的一些應用有了簡單的瞭解，寫小白篇的目的就是為初學者建立一個基本概念，讓初學者在學習的道路上建立一定的基礎。
從今天開始，我會持續更新幾篇Spring Cloud的進階教程。
Eureka簡介
Eureka是Netflix開發的服務發現框架，本身就是一個基於REST的服務。Spring Cloud將它整合在其子專案spring-cloud-netflix中，用來實現服務的註冊與發現功能。
Eureka總體架構圖
Eureka元件介紹
- 服務註冊中心叢集
分別部署在IDC1、IDC2、IDC3中心
- 服務提供者
服務提供者一個部署在IDC1，一個部署在IDC3
- 服務消費者
服務消費者一個部署在IDC1，一個部署在IDC2
元件之間的呼叫關係
服務提供者
- 啟動服務：服務提供者會向服務註冊中心發起Register請求，註冊服務。
- 執行過程中：服務提供者會定時向註冊中心傳送Renew心跳，告訴它“我還活著”。
- 停止服務提供：服務提供者會向服務註冊中心傳送Cancel請求，告訴它清空當前服務註冊資訊。
服務消費者
- 啟動後：從服務註冊中心拉取服務註冊資訊。
- 執行過程中：定時更新服務註冊資訊。
- 發起遠端呼叫：
- - 服務消費者會從服務註冊中心選擇同機房的服務提供者，然後發起遠端呼叫，只有同機房的服務提供者宕機才會去選擇其他機房的服務提供者。
- 如果服務消費者發現同機房沒有服務提供者，則會按照負載均衡演算法 選擇其他機房的服務提供者，然後發起遠端呼叫。
註冊中心
- 啟動後：從其他節點拉取服務註冊資訊
- 執行過程中：
- - 定時執行Evict任務，定時清理沒有按時傳送Renew的服務提供者，這裡的清理會將非常正常停止、網路異常等其他因素引起的所有服務。
- 接收到的Register、Renew、Cancel請求，都會同步到其他的註冊中心節點。
Eureka Server會通過Register、Renew、Get Registry等介面提供服務的註冊、發現和心跳檢測等。
Eureka Client是一個java客戶端，用於簡化與Eureka Server的互動，客戶端本身也內建了負載均衡器（預設使用round-robin方式），在啟動後會向Eureka Server傳送心跳檢測，預設週期為30s，Eureka Server如果在多個心跳週期內沒有接收到Eureka client的某一個節點的心跳請求，Eureka Server會從服務註冊中心清理到對應的Eureka Client的服務節點（預設90s）。
資料結構
服務儲存的資料結構可以簡單的理解為是一個兩層的HashMap結構（為了保證執行緒安全使用的ConcurrentHashMap），具體的我們可以檢視原始碼中的AbstractInstanceRegistry類：
private final ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>> registry = new ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>();
第一層ConcurrentHashMap的key=spring.application.name，也就是應用名稱，value為ConcurrentHashMap。
第二層ConcurrentHashMap的key=instanceId，也就是服務的唯一例項id，value為Lease物件，也就是具體的服務。Lease其實就是對InstanceInfo的包裝，裡面儲存了例項資訊、服務註冊的時間等。具體的我們可以檢視InstanceInfo原始碼。
資料儲存過程
Eureka是通過REST介面對外提供服務的。
這裡我以註冊為例（ApplicationResource），首先將PeerAwareInstanceRegistry的例項注入到ApplicationResource的成員變數的registry裡。
- ApplicationResource接收到請求後，對呼叫registry.register()方法。
@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
}
- AbstractInstanceRegistry在register方法裡完成對服務資訊的儲存。
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.expectedNumberOfClientsSendingRenews > 0) {
// Since the client wants to register it, increase the number of clients sending renews
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);
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();
}
}
從原始碼中不難看出儲存的資料結構是雙層的HashMap。
Eureka還實現了二級快取來保證即將對外傳輸的服務資訊，
一級快取：本質還是HashMap，沒有過期時間，儲存服務資訊的對外輸出的資料結構。
private final ConcurrentMap<Key, Value> readOnlyCacheMap = new ConcurrentHashMap<Key, Value>();二級快取：是guava的快取，包含失效機制，儲存服務資訊的對外輸出的資料結構。
private final LoadingCache<Key, Value> readWriteCacheMap;快取的更新：
刪除二級快取：
- client端傳送register、renew、cancel請求並更新register登錄檔之後會刪除二級快取；
server端自身的Evict任務剔除服務後會刪除二級快取；
二級快取本事設定的失效機制（指的是guava實現的readWriteCacheMap），
載入二級快取：
- client傳送Get registry請求後，如果二級快取中沒有，就會觸發guava的load機制，從registry中獲取原始的服務資訊後進行加工處理，然後放入二級快取中；
server端更新一級快取的時候，如果二級快取沒有資料也會觸發guava的load機制；
更新一級快取：
- server端內建了一個time task會定時將二級快取中的資料同步到一級快取中，這其中包括了刪除和更新。
快取的機制可以檢視ResponseCacheImpl原始碼。
Eureka的資料結構簡單總結為：
服務註冊機制
服務註冊中心、服務提供者、服務消費者在啟動後都會向服務註冊中心發起註冊服務的請求（前提是配置了註冊服務）。
註冊中心接到register請求後：
將服務資訊儲存到registry中；
更新佇列，將該事件新增到更新佇列中，給Eureka client增量同步服務資訊使用；
清空二級快取，用於保證資料的一致性；（即清空的是：readWriteCacheMap）
更新閾值；
同步服務資訊；
服務續約
服務註冊後，要定時傳送續約請求（心跳檢查），證明我還活著，不要清空我的服務資訊，定時時間預設30s，可以通過配置：eureka.instance.lease-renewal-interval-in-seconds來修改。
註冊中心接收到續約請求後（renew）：
- 更新服務物件的最近續約時間（lastUpdateTimestamp）；
- 將資訊同步給其他的節點；
服務登出
正常的服務停止之前會發送登出服務請求，通知註冊中心我要下線了。
註冊中心接收到登出請求後（cancel）：
- 將服務資訊從registry中刪除；
- 更新佇列；
- 清空二級快取；
- 更新閾值；
- 同步資訊給其他節點；
說明：只有服務正常停止才會傳送cancel請求，非正常停止的會通過Eureka Server的主動剔除機制進行刪除。
服務剔除
服務剔除其實是一個兜底的方案，目的就是解決非正常情況下的服務宕機或其他因素導致不能傳送cancel請求的服務資訊清理的策略。
服務剔除分為：
- 判斷剔除條件
- 找出過期服務
- 清理過期服務
剔除條件：
- 關閉自我保護
- 自我保護如果開啟，會先判斷是server還是client出現問題，如果是client的問題就會進行刪除；
自我保護機制：Eureka的自我保護機制是為了防止誤殺服務提供的一種保護機制。Eureka的自我保護機制認為如果有大量的服務都續約失敗，則認為自己出現了問題（例如：自己斷網了），也就不剔除了。反之，則是它人的問題，就進行剔除。
自我保護的閾值分為server和client，如果超出閾值就是表示大量服務可用，部分服務不可用，這判定為client端出現問題。如果未超出閾值就是表示大量服務不可用，則判定是自己出現了問題。
閾值的計算：
- 自我保護閾值 = 服務總數 * 每分鐘續約數 * 自我保護閾值因子；
- 每分鐘續約數 = （60s / 客戶端續約時間）；
過期服務：
找出過期服務會遍歷所有的服務，判斷上次續約時間距離當前時間大於閾值就標記為過期，同時會將這些過期的服務儲存的過期的服務集合中。
剔除服務：
剔除服務之前會先計算要是剔除的服務數量，然後遍歷過期服務，通過洗牌演算法確保每次都公平的選擇出要剔除的服務，然後進行剔除。
執行剔除服務後：
- 從register中刪除服務資訊；
- 更新佇列；
- 清空二級快取，保證資料的一致性；
服務獲取
Eureka Client服務的獲取都是從快取中獲取，如果快取中沒有，就載入資料到快取中，然後在從快取中取。服務的獲取方式分為全量同步和增量同步兩種。
registry中只儲存資料結構，快取中存ready的服務資訊
- 先讀取一級快取
- 先判斷是否開啟一級快取
- 如果開啟一級快取，就從一級快取中取，如果一級快取中沒有，則從二級快取中取；
- 如果沒有開啟一級快取，則直接從二級快取中取；
- 再讀取二級快取
- 如果二級快取中存在，則直接返回；
- 如果二級快取中不存在，則先將資料載入到二級快取中，然後再讀取二級快取中的資料。
注意：載入二級快取的時候需要判斷是全量還是增量，如果是增量的話，就從recentlyChangedQueue中載入，如果是全量的話就從registry中載入。
服務同步
服務同步是Server節點之間的資料同步。分為啟動時同步，執行時同步。
- 啟動同步
啟動同步時，會先遍歷Applications中獲取的服務資訊，並將服務資訊註冊到registry中。可以參考PeerAwareInstanceRegistryImpl類中的syncUp方法：
public int syncUp() {
// Copy entire entry from neighboring DS node
int count = 0;
for (int i = 0; ((i < serverConfig.getRegistrySyncRetries()) && (count == 0)); i++) {
if (i > 0) {
try {
Thread.sleep(serverConfig.getRegistrySyncRetryWaitMs());
} catch (InterruptedException e) {
logger.warn("Interrupted during registry transfer..");
break;
}
}
Applications apps = eurekaClient.getApplications();
for (Application app : apps.getRegisteredApplications()) {
for (InstanceInfo instance : app.getInstances()) {
try {
if (isRegisterable(instance)) {
register(instance, instance.getLeaseInfo().getDurationInSecs(), true);
count++;
}
} catch (Throwable t) {
logger.error("During DS init copy", t);
}
}
}
}
return count;
}注意這個方法使用類兩層for迴圈，第一次迴圈時保證自己已經拉取到服務資訊，第二層迴圈是遍歷拉取到服務註冊資訊。
- 執行時同步
server端當有reigster、renew、cancel請求進來時，會將這些請求封裝到一個task中，然後放到一個隊列當中，然後經過一系列的處理後，在放到另一個佇列中。 可以檢視PeerAwareInstanceRegistryImpl類中的BatchWorkerRunnable類，這裡就不再貼原始碼了。
總結
Eureka的原理接介紹到這裡，從整體上看似簡單，但實現細節相關複雜。得多看幾遍原始碼才能猜透他們的設計思路。
Eureka作為服務的註冊與發現，它實際的設計原則是遵循AP原則，也就是“資料的最終一致性”。現在還有好多公司使用zk、nacos來作為服務的註冊中心，後續會簡單更新一篇關於服務註冊中心的對比，這裡就不過多闡述。
- 寫作不易，轉載請註明出處，喜歡的小夥伴可以關注公眾號檢視更多喜歡的文章。
- 聯絡方式：[email protected] QQ:95472323