Eureka源码分析（2.1.4.Release）
首先源码切忌一行一行debug，需先了解eureka主要功能后，再分析其功能如何实现。
大家一定有疑问，eureka(一)-功能介绍与客户端之服务获取分析了客户端是通过发起restful请求给注册中心来获取服务列表的，那么注册中心即eureka服务端的服务列表数据是如何存储的？又是如何返回给客户端的？
eureka服务端的相关bean初始化
从EurekaServerAutoConfiguration出发（为什么会加载该配置类的bean？详情请看springboot自动装载）
重点分析PeerAwareInstanceRegistry和EurekaServerContext。
PeerAwareInstanceRegistry的实现类为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实例
@Bean
public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs,
PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) {
return new DefaultEurekaServerContext(this.eurekaServerConfig, serverCodecs,
registry, peerEurekaNodes, this.applicationInfoManager);
}
再来看看EurekaServerContext上下文的初始化：
com.netflix.eureka.DefaultEurekaServerContext#initialize：
@PostConstruct
@Override
public void initialize() {
logger.info("Initializing ...");
//eureka服务端节点更新任务开启
peerEurekaNodes.start();
try {
//注册中心初始化
registry.init(peerEurekaNodes);
} catch (Exception e) {
throw new RuntimeException(e);
}
logger.info("Initialized");
}
重点看看registry.init(peerEurekaNodes)的代码逻辑
com.netflix.eureka.registry.PeerAwareInstanceRegistryImpl#init：
@Override
public void init(PeerEurekaNodes peerEurekaNodes) throws Exception {
this.numberOfReplicationsLastMin.start();
this.peerEurekaNodes = peerEurekaNodes;
//重点看这里，初始化response缓存
initializedResponseCache();
//开启续约阈值更新任务
scheduleRenewalThresholdUpdateTask();
initRemoteRegionRegistry();
try {
Monitors.registerObject(this);
} catch (Throwable e) {
logger.warn("Cannot register the JMX monitor for the InstanceRegistry :", e);
}
}
初始化流程图如下：
response缓存结构
终于到了本章的重点，接上文提到的initializedResponseCache()方法：初始化response缓存结构。
@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缓存，默认为true
this.shouldUseReadOnlyResponseCache = serverConfig.shouldUseReadOnlyResponseCache();
//上文初始化的InstanceRegistry
this.registry = registry;
//responseCacheUpdateIntervalMs=30*1000,默认为30s
long responseCacheUpdateIntervalMs = serverConfig.getResponseCacheUpdateIntervalMs();
//读写缓存map，该map的结构为google的guava cache，暂不了解其原理。从方法中可大概猜测其作用
this.readWriteCacheMap =
CacheBuilder.newBuilder().initialCapacity(serverConfig.getInitialCapacityOfResponseCache())
.expireAfterWrite(serverConfig.getResponseCacheAutoExpirationInSeconds(), TimeUnit.SECONDS)//写入后，默认180s后过期
.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;
}
});
//默认为true
if (shouldUseReadOnlyResponseCache) {
//responseCacheUpdateIntervalMs=30，默认每隔30s执行一次getCacheUpdateTask()
timer.schedule(getCacheUpdateTask(),
new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
+ responseCacheUpdateIntervalMs),
responseCacheUpdateIntervalMs);
}
……忽略下半部分代码……
先来分析一下readWriteCacheMap的作用
- 写入180s后，元素过期。
- 通过generatePayload(key)生成value值。
下面再来看看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是通过registry本地注册表获取到的（registry也是一个本地缓存）。即这里可以分析到，eureka有两层缓存，上层为读写缓存map，底层为registry注册表缓存。
跳出到ResponseCacheImpl初始化中的getCacheUpdateTask方法，从字面意思是更新缓存，那么它具体的实现逻辑是什么呢？
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对应的value值赋值到只读map里面,即readOnlyCacheMap定期从readWriteMap中更新value值。
response缓存初始化流程如下：
总结：从这里可以分析到eureka注册中心的response缓存一共有3层缓存，第一层为只读缓存，第二层为读写缓存，第三层为registry本地注册表缓存。只读缓存每30s拉取读写缓存的值，读写缓存写入180s后过期，如果要获取的key没有value值时，则通过registry注册表缓存获取数据。
response缓存结构是如何实现读功能的？
response缓存主要作用于客户端与eureka注册中心交互的时候。
从客户端向注册中心获取服务列表的功能中，可以分析出response缓存是如何实现读功能的。
获取服务列表时，服务端的运行流程如下：
- 默认读取只读map，如果只读map没有，则读取读写map，如果读写map也没有，就读取registry本地注册表缓存。
- registry本地注册表存储的是最新的服务列表数据。（registry的具体存储逻辑暂不深究）
问题：为什么这样设计？
这让我想起主从数据库的读写分离，数据库的读写分离是为了分摊主数据库服务器的读写压力。而eureka所设计的缓存级别无疑也是为了读写分离，因为在写的时候，如ConcurrentHashmap会持有桶节点对象的锁，阻塞同一个桶的读写线程。这样设计的话，线程在写的时候，并不会影响读操作，避免了争抢资源所带来的压力。
问题：该缓存结构如何保证最终一致性？
- 从只读map中获取key对应的值，如果只读map没有value值的时候，会从读写缓存里面获取，而读写缓存180s后过期，所以，它又会从本地注册表中获取到最新的实例信息。
- 只读map中会每30s遍历，将读写map里面的key赋值到只读map中。
问题：如果有新的实例注册，极端情况下难道要等读写缓存的key，180s后过期，才能获取到最新的服务列表数据吗？即在实例有变化的时候，服务端又是如何实现的？
服务端接受客户端注册
带着疑问，我们再来分析一下，服务端是如何实现客户端的注册操作的。
具体流程如下：（可以根据流程所提及到的方法进行分析，这里不把代码贴出来了）
结论：在接受客户端注册的时候，服务端会将读写缓存的key清掉，30s后只读缓存从读写缓存拉取数据的时候，该服务列表获取到的是最新的数据。如果客户端下线，同样地，读写缓存也会被清除掉。所以极端情况，最长30s后，客户端才能获取到最新的服务列表。
优点：
- 尽可能保证内存注册表数据不会出现频繁的读写冲突
- 保证对eureka服务端的请求读取的都是内存，性能高。
在以后的开发工作中，面对频繁的读写资源争抢的情况，也可以考虑采用多级缓存这种方案来设计系统。
