一、摘要
本文章将继续深入了解eureka的复制算法。
这里先分析eureka-server启动的时候如何初始化数据。
二、分析
eureka-server启动的时候会尝试从其他eureka结点获取应用信息。
从图上我们可以看到，eureka-server其实本身也是eureka-client，跟其他结点的数据同步会使用DiscoveryClient同步数据到其他结点。
eureka数据初始化
1. eureka-server启动
EurekaBootStrap会调用PeerAwareInstanceRegistryImpl.syncUp方法
image.png
PeerAwareInstanceRegistryImpl是eureka节点复制的核心类。
/**
* Handles replication of all operations to {@link AbstractInstanceRegistry} to peer
* <em>Eureka</em> nodes to keep them all in sync.
*
* <p>
* Primary operations that are replicated are the
* <em>Registers,Renewals,Cancels,Expirations and Status Changes</em>
* </p>
*
* <p>
* When the eureka server starts up it tries to fetch all the registry
* information from the peer eureka nodes.If for some reason this operation
* fails, the server does not allow the user to get the registry information for
* a period specified in
* {@link com.netflix.eureka.EurekaServerConfig#getWaitTimeInMsWhenSyncEmpty()}.
* </p>
*
* <p>
* One important thing to note about <em>renewals</em>.If the renewal drops more
* than the specified threshold as specified in
* {@link com.netflix.eureka.EurekaServerConfig#getRenewalPercentThreshold()} within a period of
* {@link com.netflix.eureka.EurekaServerConfig#getRenewalThresholdUpdateIntervalMs()}, eureka
* perceives this as a danger and stops expiring instances.
* </p>
*
* @author Karthik Ranganathan, Greg Kim
*
*/
复制算法的核心逻辑统一在PeerAwareInstanceRegistryImpl上实现。
PeerAwareInstanceRegistryImpl类图
2. PeerAwareInstanceRegistryImpl.syncUp
/**
* Populates the registry information from a peer eureka node. This
* operation fails over to other nodes until the list is exhausted if the
* communication fails.
*/
@Override
public int syncUp() {
logger.info("PeerAwareInstanceRegistryImpl.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;
}
我们可以简单看到，通过syncUp的方法，eureka会通过eurekaClient，把其他eureka-server结点的数据同步注册到本地(内存)。
3. DiscoveryClient
DiscoveryClient是用来跟eurekaServer通讯的桥梁，但是DiscoveryClient并不是简单地从eurekaServer获取实时的数据，其实DiscoveryClient本身也有做缓存。
DiscoveryClient类图
/**
* Gets the full registry information from the eureka server and stores it locally.
* When applying the full registry, the following flow is observed:
*
* if (update generation have not advanced (due to another thread))
* atomically set the registry to the new registry
* fi
*
* @return the full registry information.
* @throws Throwable
* on error.
*/
//拉取eureka server的数据到本地内存
private void getAndStoreFullRegistry() throws Throwable {
long currentUpdateGeneration = fetchRegistryGeneration.get();
logger.info("Getting all instance registry info from the eureka server");
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());
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");
}
}
@Override
public Applications getApplications() {
return localRegionApps.get();
}
/**
* The task that fetches the registry information at specified intervals.
*
*/
class CacheRefreshThread implements Runnable {
public void run() {
refreshRegistry();
}
}
/**
* Initializes all scheduled tasks.
*/
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,
//拉起一个定时任务定期拉取eureka-server的数据到本地
new CacheRefreshThread()
),
registryFetchIntervalSeconds, TimeUnit.SECONDS);
}
我截取了核心的源码，主要的逻辑就是initScheduledTasks会起一个定时任务，定期拉取eureka-server的数据到本地。所以我们可以直观地理解DiscoveryClient的数据也并不是实时的。
三、总结
这篇文章分析了eureka-server的数据初始化流程，那么当结点的数据更新后如何把这部分数据复制到其他结点？后面的文章会继续分析
网友评论