源码流程图
先上图，不太清晰，抱歉
一、Eureka Server源码分析
/**
* 激活Eureka服务器相关配置的注释
* Annotation to activate Eureka Server related configuration {@link EurekaServerAutoConfiguration}
*
* @author Dave Syer
* @author Biju Kunjummen
*
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(EurekaServerMarkerConfiguration.class)
public @interface EnableEurekaServer {
}
- 从注解可以看到它@link 了配置启动类EurekaServerAutoConfiguration，这个类会将配置信息读取进来
@Configuration
@Import(EurekaServerInitializerConfiguration.class)
@ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class)
@EnableConfigurationProperties({ EurekaDashboardProperties.class,InstanceRegistryProperties.class })
@PropertySource("classpath:/eureka/server.properties")
public class EurekaServerAutoConfiguration extends WebMvcConfigurerAdapter {
}
-
EurekaServerAutoConfiguration上有一个注解@Import(EurekaServerInitializerConfiguration.class)，@Import的作用会将这个类的实例加入IOC容器。
-
所以EurekaServerInitializerConfiguration这个类会被初始化，它实现了SmartLifecycle接口，当ApplicationContext自身启动时会调用他的start()方法。
@Configuration
public class EurekaServerInitializerConfiguration
implements ServletContextAware,SmartLifecycle,Ordered {
//省略部分代码
@Override
public void start() {
new Thread(new Runnable() {
@Override
public void run() {
try {
//Todo: is this class even needed Now?
//初始化EurekaServer，同时启动Eureka Server
eurekaServerBootstrap.contextinitialized(EurekaServerInitializerConfiguration.this.servletContext);
log.info("Started Eureka Server");
//事件发布
publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));
EurekaServerInitializerConfiguration.this.running = true;
publish(new EurekaServerStartedEvent(getEurekaServerConfig()));
}
catch (Exception ex) {
// Help!
log.error("Could not initialize Eureka servlet context",ex);
}
}
}).start();
}
}
public void contextinitialized(ServletContext context) {
try {
//初始化环境变量
initEurekaEnvironment();
//初始化上下文
initEurekaServerContext();
context.setAttribute(EurekaServerContext.class.getName(),this.serverContext);
}
catch (Throwable e) {
log.error("Cannot bootstrap eureka server :",e);
throw new RuntimeException("Cannot bootstrap eureka server :",e);
}
}
- initEurekaEnvironment里面都是一些环境变量初始化，不用太关心。主要看initEurekaServerContext()这个初始化上下文的方法
protected void initEurekaServerContext() throws Exception {
// For backward compatibility
jsonxStream.getInstance().registerConverter(new V1AwareInstanceInfoConverter(),XStream.PRIORITY_VERY_HIGH);
XmlXStream.getInstance().registerConverter(new V1AwareInstanceInfoConverter(),XStream.PRIORITY_VERY_HIGH);
if (isAws(this.applicationInfoManager.getInfo())) {
this.awsBinder = new AwsBinderDelegate(this.eurekaServerConfig,this.eurekaClientConfig,this.registry,this.applicationInfoManager);
this.awsBinder.start();
}
EurekaServerContextHolder.initialize(this.serverContext);
log.info("Initialized server context");
// copy registry from neighboring eureka node
int registryCount = this.registry.syncUp();
//这句重要，里面会打开定时器
this.registry.openForTraffic(this.applicationInfoManager,registryCount);
// Register all monitoring statistics.
//注册所有监视统计信息
Eurekamonitors.registerallStats();
}
- 进入openForTraffic方法内部
@Override
public void openForTraffic(ApplicationInfoManager applicationInfoManager,int count) {
// renewals happen every 30 seconds and for a minute it should be a factor of 2.
this.expectednumberOfRenewsPerMin = count * 2;
this.numberOfRenewsPerMinThreshold =
(int) (this.expectednumberOfRenewsPerMin * serverConfig.getrenewalPercentThreshold());
logger.info("Got {} instances from neighboring DS node",count);
logger.info("Renew threshold is: {}",numberOfRenewsPerMinThreshold);
this.startupTime = System.currentTimeMillis();
if (count > 0) {
this.peerInstancesTransferEmptyOnStartup = false;
}
DataCenterInfo.Name selfName = applicationInfoManager.getInfo().getDataCenterInfo().getName();
boolean isAws = Name.Amazon == selfName;
if (isAws && serverConfig.shouldPrimeAwsReplicaConnections()) {
logger.info("Priming AWS connections for all replicas..");
primeAwsReplicas(applicationInfoManager);
}
logger.info("Changing status to UP");
applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
//启动定时器
super.postinit();
}
protected void postinit() {
renewsLastMin.start();
if (evictionTaskRef.get() != null) {
evictionTaskRef.get().cancel();
}
evictionTaskRef.set(new evictionTask());
evictionTimer.schedule(evictionTaskRef.get(),serverConfig.getevictionIntervalTimerInMs(),serverConfig.getevictionIntervalTimerInMs());
}
- 剔除服务的方法可以看看，主要思想就是找到要剔除服务的服务id，然后从map移除
public void evict(long additionalLeaseMs) {
logger.debug("Running the evict task");
if (!isLeaseExpirationEnabled()) {
logger.debug("DS: lease expiration is currently disabled.");
return;
}
// We collect first all expired items,to evict them in random order. For large eviction sets,// if we do not that,we might wipe out whole apps before self preservation kicks in. By randomizing it,// the impact should be evenly distributed across all applications.
List<Lease<InstanceInfo>> expiredLeases = new ArrayList<>();
for (Entry<String,Map<String,Lease<InstanceInfo>>> groupEntry : registry.entrySet()) {
Map<String,Lease<InstanceInfo>> leaseMap = groupEntry.getValue();
if (leaseMap != null) {
for (Entry<String,Lease<InstanceInfo>> leaseEntry : leaseMap.entrySet()) {
Lease<InstanceInfo> lease = leaseEntry.getValue();
if (lease.isExpired(additionalLeaseMs) && lease.getHolder() != null) {
expiredLeases.add(lease);
}
}
}
}
// To compensate for GC pauses or drifting local time,we need to use current registry size as a base for
// triggering self-preservation. Without that we would wipe out full registry.
int registrySize = (int) getLocalRegistrySize();
int registrySizeThreshold = (int) (registrySize * serverConfig.getrenewalPercentThreshold());
int evictionLimit = registrySize - registrySizeThreshold;
int toevict = Math.min(expiredLeases.size(),evictionLimit);
if (toevict > 0) {
logger.info("evicting {} items (expired={},evictionLimit={})",toevict,expiredLeases.size(),evictionLimit);
Random random = new Random(System.currentTimeMillis());
for (int i = 0; i < toevict; i++) {
// Pick a random item (knuth shuffle algorithm)
int next = i + random.nextInt(expiredLeases.size() - i);
Collections.swap(expiredLeases,i,next);
Lease<InstanceInfo> lease = expiredLeases.get(i);
String appName = lease.getHolder().getAppName();
String id = lease.getHolder().getId();
EXPIRED.increment();
logger.warn("DS: Registry: expired lease for {}/{}",appName,id);
//根据id移除
internalCancel(appName,id,false);
}
}
}
二、Eureka客户端源码
通过看源码想要解决的问题 ：
@Bean(destroyMethod = "shutdown")
@ConditionalOnMissingBean(value = EurekaClient.class,search = SearchStrategy.CURRENT)
@org.springframework.cloud.context.config.annotation.RefreshScope
@Lazy
public EurekaClient eurekaClient(ApplicationInfoManager manager,EurekaClientConfig config,EurekaInstanceConfig instance) {
manager.getInfo(); // force initialization
return new CloudEurekaClient(manager,config,this.optionalArgs,this.context);
}
try {
// default size of 2 - 1 each for heartbeat and cacheRefresh
scheduler = Executors.newScheduledThreadPool(2,new ThreadFactoryBuilder()
.setNameFormat("discoveryClient-%d")
.setDaemon(true)
.build());
heartbeatExecutor = new ThreadPoolExecutor(
1,clientConfig.getHeartbeatExecutorThreadPoolSize(),TimeUnit.SECONDS,new SynchronousQueue<Runnable>(),new ThreadFactoryBuilder()
.setNameFormat("discoveryClient-HeartbeatExecutor-%d")
.setDaemon(true)
.build()
); // use direct handoff
cacheRefreshExecutor = new ThreadPoolExecutor(
1,clientConfig.getCacheRefreshExecutorThreadPoolSize(),new ThreadFactoryBuilder()
.setNameFormat("discoveryClient-CacheRefreshExecutor-%d")
.setDaemon(true)
.build()
); // use direct handoff
eurekaTransport = new EurekaTransport();
scheduleServerEndpointTask(eurekaTransport,args);
AzToRegionMapper azToRegionMapper;
if (clientConfig.shouldUsednsForFetchingServiceUrls()) {
azToRegionMapper = new DNSBasedAzToRegionMapper(clientConfig);
} else {
azToRegionMapper = new PropertyBasedAzToRegionMapper(clientConfig);
}
if (null != remoteRegionsToFetch.get()) {
azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(","));
}
instanceRegionChecker = new InstanceRegionChecker(azToRegionMapper,clientConfig.getRegion());
} catch (Throwable e) {
throw new RuntimeException("Failed to initialize discoveryClient!",e);
}
if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) {
fetchRegistryFromBackup();
}
// call and execute the pre registration handler before all background tasks (inc registration) is started
if (this.preRegistrationHandler != null) {
this.preRegistrationHandler.beforeRegistration();
}
if (clientConfig.shouldRegisterWithEureka() && clientConfig.shouldEnforceRegistrationAtinit()) {
try {
if (!register() ) {
throw new IllegalStateException("Registration error at startup. Invalid server response.");
}
} catch (Throwable th) {
logger.error("Registration error at startup: {}",th.getMessage());
throw new IllegalStateException(th);
}
}
// finally,init the schedule tasks (e.g. cluster resolvers,heartbeat,instanceInfo replicator,fetch
//初始化定时器任务
initScheduledTasks();
try {
Monitors.registerObject(this);
} catch (Throwable e) {
logger.warn("Cannot register timers",e);
}
- 在initScheduledTasks方法里有一个状态改变监听器Statuschangelistener，他会触发instanceInfoReplicator.ondemandUpdate();方法，该方法会调用run方法
- 在InstanceInfoReplicator的run方法里调用了 discoveryClient.register()方法
- 就是该方法完成了服务注册，看下面代码可以看出通过http方式请求服务端完成了注册
boolean register() throws Throwable {
logger.info(PREFIX + "{}: registering service...",appPathIdentifier);
EurekaHttpResponse<Void> httpResponse;
try {
httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
} catch (Exception e) {
logger.warn(PREFIX + "{} - registration Failed {}",appPathIdentifier,e.getMessage(),e);
throw e;
}
if (logger.isInfoEnabled()) {
logger.info(PREFIX + "{} - registration status: {}",httpResponse.getStatusCode());
}
return httpResponse.getStatusCode() == 204;
}
- 可以看到客户端有日志{registering service...}打印，说明确实执行到了这个方法。
- 并且会发现有两个定时任务一直在执行，一个HeartbeatThread，一个CacheRefreshThread
多级缓存思想
- 在拉取注册表的时候
- 注册表发生变更的时候
- 会在内存中更新变更的注册表数据，同时过期掉ReadWriteCacheMap。
- 此过程中不会影响ReadOnlyCacheMap提供使用。
- 默认30秒ReadWriteCacheMap的数据更新到ReadOnlyCacheMap
- 默认180秒秒ReadWriteCacheMap的数据失效
- 下次有服务拉取注册表，又会从内存中获取最新数据了，同时填充各级缓存
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 [email protected] 举报，一经查实，本站将立刻删除。
