Eureka源碼分析（2.1.4.Release）
首先源碼切忌一行一行debug，需先了解eureka主要功能后咙咽，再分析其功能如何實(shí)現(xiàn)洪燥。
大家一定有疑問(wèn)，eureka(一)-功能介紹與客戶端之服務(wù)獲取分析了客戶端是通過(guò)發(fā)起restful請(qǐng)求給注冊(cè)中心來(lái)獲取服務(wù)列表的脐湾，那么注冊(cè)中心即eureka服務(wù)端的服務(wù)列表數(shù)據(jù)是如何存儲(chǔ)的臭笆？又是如何返回給客戶端的？
eureka服務(wù)端的相關(guān)bean初始化
從EurekaServerAutoConfiguration出發(fā)（為什么會(huì)加載該配置類(lèi)的bean秤掌？詳情請(qǐng)看springboot自動(dòng)裝載）
重點(diǎn)分析PeerAwareInstanceRegistry和EurekaServerContext愁铺。
PeerAwareInstanceRegistry的實(shí)現(xiàn)類(lèi)為InstanceRegistry：
@Bean
public PeerAwareInstanceRegistry peerAwareInstanceRegistry(
ServerCodecs serverCodecs) {
this.eurekaClient.getApplications(); // force initialization
return new InstanceRegistry(this.eurekaServerConfig, this.eurekaClientConfig,
serverCodecs, this.eurekaClient,
this.instanceRegistryProperties.getExpectedNumberOfClientsSendingRenews(),
this.instanceRegistryProperties.getDefaultOpenForTrafficCount());
}
EurekaServerContext初始化，引入了上面提到的PeerAwareInstanceRegistry實(shí)例
@Bean
public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs,
PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) {
return new DefaultEurekaServerContext(this.eurekaServerConfig, serverCodecs,
registry, peerEurekaNodes, this.applicationInfoManager);
}
再來(lái)看看EurekaServerContext上下文的初始化：
com.netflix.eureka.DefaultEurekaServerContext#initialize：
@PostConstruct
@Override
public void initialize() {
logger.info("Initializing ...");
//eureka服務(wù)端節(jié)點(diǎn)更新任務(wù)開(kāi)啟
peerEurekaNodes.start();
try {
//注冊(cè)中心初始化
registry.init(peerEurekaNodes);
} catch (Exception e) {
throw new RuntimeException(e);
}
logger.info("Initialized");
}
重點(diǎn)看看registry.init(peerEurekaNodes)的代碼邏輯
com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl#init：
@Override
public void init(PeerEurekaNodes peerEurekaNodes) throws Exception {
this.numberOfReplicationsLastMin.start();
this.peerEurekaNodes = peerEurekaNodes;
//重點(diǎn)看這里闻鉴，初始化response緩存
initializedResponseCache();
//開(kāi)啟續(xù)約閾值更新任務(wù)
scheduleRenewalThresholdUpdateTask();
initRemoteRegionRegistry();
try {
Monitors.registerObject(this);
} catch (Throwable e) {
logger.warn("Cannot register the JMX monitor for the InstanceRegistry :", e);
}
}
初始化流程圖如下：
response緩存結(jié)構(gòu)
終于到了本章的重點(diǎn)茵乱，接上文提到的initializedResponseCache()方法：初始化response緩存結(jié)構(gòu)。
@Override
public synchronized void initializedResponseCache() {
if (responseCache == null) {
responseCache = new ResponseCacheImpl(serverConfig, serverCodecs, this);
}
}
ResponseCacheImpl(EurekaServerConfig serverConfig, ServerCodecs serverCodecs, AbstractInstanceRegistry registry) {
this.serverConfig = serverConfig;
this.serverCodecs = serverCodecs;
//是否使用只讀模式的response緩存孟岛，默認(rèn)為true
this.shouldUseReadOnlyResponseCache = serverConfig.shouldUseReadOnlyResponseCache();
//上文初始化的InstanceRegistry
this.registry = registry;
//responseCacheUpdateIntervalMs=30*1000,默認(rèn)為30s
long responseCacheUpdateIntervalMs = serverConfig.getResponseCacheUpdateIntervalMs();
//讀寫(xiě)緩存map瓶竭，該map的結(jié)構(gòu)為google的guava cache，暫不了解其原理渠羞。從方法中可大概猜測(cè)其作用
this.readWriteCacheMap =
CacheBuilder.newBuilder().initialCapacity(serverConfig.getInitialCapacityOfResponseCache())
.expireAfterWrite(serverConfig.getResponseCacheAutoExpirationInSeconds(), TimeUnit.SECONDS)//寫(xiě)入后斤贰，默認(rèn)180s后過(guò)期
.removalListener(new RemovalListener<Key, Value>() {
@Override
public void onRemoval(RemovalNotification<Key, Value> notification) {
Key removedKey = notification.getKey();
if (removedKey.hasRegions()) {
Key cloneWithNoRegions = removedKey.cloneWithoutRegions();
regionSpecificKeys.remove(cloneWithNoRegions, removedKey);
}
}
})
//加載key的value值
.build(new CacheLoader<Key, Value>() {
@Override
public Value load(Key key) throws Exception {
if (key.hasRegions()) {
Key cloneWithNoRegions = key.cloneWithoutRegions();
regionSpecificKeys.put(cloneWithNoRegions, key);
}
//value值生成
Value value = generatePayload(key);
return value;
}
});
//默認(rèn)為true
if (shouldUseReadOnlyResponseCache) {
//responseCacheUpdateIntervalMs=30，默認(rèn)每隔30s執(zhí)行一次getCacheUpdateTask()
timer.schedule(getCacheUpdateTask(),
new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
+ responseCacheUpdateIntervalMs),
responseCacheUpdateIntervalMs);
}
……忽略下半部分代碼……
先來(lái)分析一下readWriteCacheMap的作用
- 寫(xiě)入180s后次询，元素過(guò)期荧恍。
- 通過(guò)generatePayload(key)生成value值。
下面再來(lái)看看generatePayload(key)又是如何生成value的。如傳入key為“ALL_APPS”
private Value generatePayload(Key key) {
Stopwatch tracer = null;
try {
String payload;
switch (key.getEntityType()) {
case Application:
boolean isRemoteRegionRequested = key.hasRegions();
if (ALL_APPS.equals(key.getName())) {
if (isRemoteRegionRequested) {
tracer = serializeAllAppsWithRemoteRegionTimer.start();
payload = getPayLoad(key, registry.getApplicationsFromMultipleRegions(key.getRegions()));
} else {
tracer = serializeAllAppsTimer.start();
//debug模式下送巡，可知跑到這里獲取value值摹菠。
payload = getPayLoad(key, registry.getApplications());
}
}
……忽略下部分代碼……
}
可知readWriteCacheMap的key是通過(guò)registry本地注冊(cè)表獲取到的（registry也是一個(gè)本地緩存）。即這里可以分析到骗爆，eureka有兩層緩存次氨，上層為讀寫(xiě)緩存map，底層為registry注冊(cè)表緩存摘投。
跳出到ResponseCacheImpl初始化中的getCacheUpdateTask方法煮寡，從字面意思是更新緩存，那么它具體的實(shí)現(xiàn)邏輯是什么呢谷朝？
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()) {
logger.debug("Updating the client cache from response cache for key : {} {} {} {}",
key.getEntityType(), key.getName(), key.getVersion(), key.getType());
}
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 for key {}", key.toStringCompact(), th);
}
}
}
};
}
遍歷只讀map(readOnlyCacheMap)中的key洲押，將readWriteMap對(duì)應(yīng)的value值賦值到只讀map里面,即readOnlyCacheMap定期從readWriteMap中更新value值。
response緩存初始化流程如下：
總結(jié)：從這里可以分析到eureka注冊(cè)中心的response緩存一共有3層緩存圆凰，第一層為只讀緩存杈帐，第二層為讀寫(xiě)緩存，第三層為registry本地注冊(cè)表緩存专钉。只讀緩存每30s拉取讀寫(xiě)緩存的值挑童，讀寫(xiě)緩存寫(xiě)入180s后過(guò)期，如果要獲取的key沒(méi)有value值時(shí)跃须，則通過(guò)registry注冊(cè)表緩存獲取數(shù)據(jù)站叼。
response緩存結(jié)構(gòu)是如何實(shí)現(xiàn)讀功能的？
response緩存主要作用于客戶端與eureka注冊(cè)中心交互的時(shí)候菇民。
從客戶端向注冊(cè)中心獲取服務(wù)列表的功能中尽楔，可以分析出response緩存是如何實(shí)現(xiàn)讀功能的。
獲取服務(wù)列表時(shí)第练，服務(wù)端的運(yùn)行流程如下：
- 默認(rèn)讀取只讀map阔馋，如果只讀map沒(méi)有，則讀取讀寫(xiě)map娇掏，如果讀寫(xiě)map也沒(méi)有呕寝，就讀取registry本地注冊(cè)表緩存。
- registry本地注冊(cè)表存儲(chǔ)的是最新的服務(wù)列表數(shù)據(jù)婴梧。（registry的具體存儲(chǔ)邏輯暫不深究）
問(wèn)題：為什么這樣設(shè)計(jì)下梢？
這讓我想起主從數(shù)據(jù)庫(kù)的讀寫(xiě)分離，數(shù)據(jù)庫(kù)的讀寫(xiě)分離是為了分?jǐn)傊鲾?shù)據(jù)庫(kù)服務(wù)器的讀寫(xiě)壓力塞蹭。而eureka所設(shè)計(jì)的緩存級(jí)別無(wú)疑也是為了讀寫(xiě)分離孽江，因?yàn)樵趯?xiě)的時(shí)候，如ConcurrentHashmap會(huì)持有桶節(jié)點(diǎn)對(duì)象的鎖浮还，阻塞同一個(gè)桶的讀寫(xiě)線程竟坛。這樣設(shè)計(jì)的話，線程在寫(xiě)的時(shí)候钧舌，并不會(huì)影響讀操作担汤，避免了爭(zhēng)搶資源所帶來(lái)的壓力。
問(wèn)題：該緩存結(jié)構(gòu)如何保證最終一致性洼冻？
- 從只讀map中獲取key對(duì)應(yīng)的值崭歧，如果只讀map沒(méi)有value值的時(shí)候，會(huì)從讀寫(xiě)緩存里面獲取撞牢，而讀寫(xiě)緩存180s后過(guò)期率碾，所以，它又會(huì)從本地注冊(cè)表中獲取到最新的實(shí)例信息屋彪。
- 只讀map中會(huì)每30s遍歷所宰，將讀寫(xiě)map里面的key賦值到只讀map中。
問(wèn)題：如果有新的實(shí)例注冊(cè)畜挥，極端情況下難道要等讀寫(xiě)緩存的key仔粥，180s后過(guò)期，才能獲取到最新的服務(wù)列表數(shù)據(jù)嗎蟹但？即在實(shí)例有變化的時(shí)候躯泰，服務(wù)端又是如何實(shí)現(xiàn)的？
服務(wù)端接受客戶端注冊(cè)
帶著疑問(wèn)华糖，我們?cè)賮?lái)分析一下麦向，服務(wù)端是如何實(shí)現(xiàn)客戶端的注冊(cè)操作的骆膝。
具體流程如下：（可以根據(jù)流程所提及到的方法進(jìn)行分析馅袁，這里不把代碼貼出來(lái)了）
結(jié)論：在接受客戶端注冊(cè)的時(shí)候，服務(wù)端會(huì)將讀寫(xiě)緩存的key清掉念颈，30s后只讀緩存從讀寫(xiě)緩存拉取數(shù)據(jù)的時(shí)候兼搏，該服務(wù)列表獲取到的是最新的數(shù)據(jù)卵慰。如果客戶端下線，同樣地向族，讀寫(xiě)緩存也會(huì)被清除掉呵燕。所以極端情況，最長(zhǎng)30s后件相，客戶端才能獲取到最新的服務(wù)列表再扭。
優(yōu)點(diǎn)：
- 盡可能保證內(nèi)存注冊(cè)表數(shù)據(jù)不會(huì)出現(xiàn)頻繁的讀寫(xiě)沖突
- 保證對(duì)eureka服務(wù)端的請(qǐng)求讀取的都是內(nèi)存，性能高夜矗。
在以后的開(kāi)發(fā)工作中泛范，面對(duì)頻繁的讀寫(xiě)資源爭(zhēng)搶的情況，也可以考慮采用多級(jí)緩存這種方案來(lái)設(shè)計(jì)系統(tǒng)紊撕。
