投稿
  1. 极客之音首页
  2. 技术随笔

【SpringCloud技术专题】「Eureka源码分析」从源码层面让你认识Eureka工作流程和运作机制（上）

小半 • • 技术随笔 • 阅读 388

导读：本篇文章讲解 【SpringCloud技术专题】「Eureka源码分析」从源码层面让你认识Eureka工作流程和运作机制（上），希望对大家有帮助，欢迎收藏，转发！站点地址：www.bmabk.com

前言介绍

了解到了SpringCloud，大家都应该知道注册中心，而对于我们从过去到现在，SpringCloud中用的最多的注册中心就是Eureka了，所以深入Eureka的原理和源码，接下来我们要进行讲解下eureka的源码分析，由此应运而产生的本章节的内容。

基本原理

  1. Eureka Server提供服务注册服务，各个节点启动后，会在Eureka Server中进行注册，这样Eureka Server中的服务注册表中将会存储所有可用服务节点的信息，服务节点的信息可以在界面中直观的看到。

  2. Eureka Client 是一个Java 客户端，用于简化与Eureka Server的交互，客户端同时也具备一个内置的、使用轮询负载算法的负载均衡器。

  3. 在应用启动后，将会向Eureka Server发送心跳(默认周期为30秒)，如果Eureka Server在多个心跳周期（默认3个周期）没有收到某个节点的心跳，Eureka Server 将会从服务注册表中把这个服务节点移除(默认90秒)。

  4. Eureka Server之间将会通过复制的方式完成数据的同步；

  5. Eureka Client具有缓存的机制，即使所有的Eureka Server都挂掉的话，客户端依然可以利用缓存中的信息消费其它服务的API；

启动流程分析

从EurekaServer 启动的流程日志入手分析：

2021-01-21 18:14:17.635  INFO 5288 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Located managed bean 'environmentManager': registering with JMX server as MBean [org.springframework.cloud.context.environment:name=environmentManager,type=EnvironmentManager]
2021-01-21 18:14:17.650  INFO 5288 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Located managed bean 'restartEndpoint': registering with JMX server as MBean [org.springframework.cloud.context.restart:name=restartEndpoint,type=RestartEndpoint]
2021-01-21 18:14:17.661  INFO 5288 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Located managed bean 'refreshScope': registering with JMX server as MBean [org.springframework.cloud.context.scope.refresh:name=refreshScope,type=RefreshScope]
2021-01-21 18:14:17.674  INFO 5288 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Located managed bean 'configurationPropertiesRebinder': registering with JMX server as MBean [org.springframework.cloud.context.properties:name=configurationPropertiesRebinder,context=335b5620,type=ConfigurationPropertiesRebinder]
2021-01-21 18:14:17.683  INFO 5288 --- [           main] o.s.j.e.a.AnnotationMBeanExporter        : Located managed bean 'refreshEndpoint': registering with JMX server as MBean [org.springframework.cloud.endpoint:name=refreshEndpoint,type=RefreshEndpoint]
2021-01-21 18:14:17.926  INFO 5288 --- [           main] o.s.c.support.DefaultLifecycleProcessor  : Starting beans in phase 0
2021-01-21 18:14:17.927  INFO 5288 --- [           main] c.n.e.EurekaDiscoveryClientConfiguration : Registering application unknown with eureka with status UP
2021-01-21 18:14:17.927  INFO 5288 --- [      Thread-10] o.s.c.n.e.server.EurekaServerBootstrap   : Setting the eureka configuration..
2021-01-21 18:14:17.948  INFO 5288 --- [      Thread-10] o.s.c.n.e.server.EurekaServerBootstrap   : isAws returned false
2021-01-21 18:14:17.949  INFO 5288 --- [      Thread-10] o.s.c.n.e.server.EurekaServerBootstrap   : Initialized server context
2021-01-21 18:14:17.949  INFO 5288 --- [      Thread-10] c.n.e.r.PeerAwareInstanceRegistryImpl    : Got 1 instances from neighboring DS node
2021-01-21 18:14:17.949  INFO 5288 --- [      Thread-10] c.n.e.r.PeerAwareInstanceRegistryImpl    : Renew threshold is: 1
2021-01-21 18:14:17.949  INFO 5288 --- [      Thread-10] c.n.e.r.PeerAwareInstanceRegistryImpl    : Changing status to UP
2021-01-21 18:14:17.958  INFO 5288 --- [      Thread-10] e.s.EurekaServerInitializerConfiguration : Started Eureka Server
2021-01-21 18:14:18.019  INFO 5288 --- [           main] s.b.c.e.t.TomcatEmbeddedServletContainer : Tomcat started on port(s): 8761 (http)
2021-01-21 18:14:18.020  INFO 5288 --- [           main] c.n.e.EurekaDiscoveryClientConfiguration : Updating port to 8761
2021-01-21 18:14:18.023  INFO 5288 --- [           main] c.s.cloud.EurekaServerApplication        : Started EurekaServerApplication in 8.299 seconds (JVM running for 8.886)

Eureka微服务已启动.

日志打印“Setting the eureka configuration..”，eureka开始进行配置，说不定也许就是Eureka Server流程启动的开始呢？我们抱着怀疑的心态进入这行日志打印的EurekaServerBootstrap类去看看。

EurekaServerBootstrap类

看看，看这个类的名字，见名知意，应该就是 EurekaServer 的启动类了：

protected void initEurekaEnvironment() throws Exception {
	log.info("Setting the eureka configuration..");
｝
initEurekaEnvironment方法

我们看到日志在initEurekaEnvironment方法中被打印出来，然后我顺着这个方法寻找该方法被调用的地方；

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);
	}
}
contextInitialized

接着发现contextInitialized这个方法里面调用了initEurekaEnvironment 方法，接着我们再往上层寻找被调用的地方；

EurekaServerInitializerConfiguration

接着我们看到 EurekaServerInitializerConfiguration 类中有个 start 方法，该方法创建了一个线程来后台执行 EurekaServer 的初始化流程；

进入 EurekaServerInitializerConfiguration 方法，看看这个所谓的 EurekaServer 初始化配置做了哪些事情？

@Override
public void start() { 
	new Thread(new Runnable() {
		@Override
		public void run() {
			try {
				//TODO: is this class even needed now?
				eurekaServerBootstrap.contextInitialized(EurekaServerInitializerConfiguration.this.servletContext);
				log.info("Started Eureka Server");
				publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));//读取相关的eureka的配置信息
				EurekaServerInitializerConfiguration.this.running = true;
				publish(new EurekaServerStartedEvent(getEurekaServerConfig())); //发布相关eureka server配置的事件操作
			}
			catch (Exception ex) {
				// Help!
				log.error("Could not initialize Eureka servlet context", ex);
			}
		}
	}).start();
}

  • 看到 log.info(“Started Eureka Server”); 这行代码，相信大家已经释然了，这里就是所谓的启动了 EurekaServer 了，其实也就是 eurekaServerBootstrap.contextInitialized(EurekaServerInitializerConfiguration.this.servletContext) 初始化了一些我们未知的东西；

  • 当打印完启动Eureka Server日志后，调用了两次 publish 方法，该方法最终调用的是this.applicationContext.publishEvent(event) 方法，目的是利用Spring中ApplicationContext对事件传递性质，事件发布者(applicationContext)来发布事件(event)，但是缺少的是监听者，其实你仔细搜索下代码，发现好像没有地方对 EurekaServerStartedEvent、EurekaRegistryAvailableEvent 进行监听。

  • 然后找到 EurekaServerStartedEvent 所在的目录下，EurekaInstanceCanceledEvent、EurekaInstanceRegisteredEvent、EurekaInstanceRenewedEvent、EurekaRegistryAvailableEvent、EurekaServerStartedEvent 有这么几个事件的类，服务下线事件、服务注册事件、服务续约事件、注册中心启动事件、Eureka Server启动事件，这么几个事件都没有被监听。

  • 像这样 @EventListener public void listen(EurekaInstanceCanceledEvent event) {下线逻辑 ｝，添加EventListener 监听注解，就可以在我们自己的代码逻辑中收到这个事件的回调了，所以想想SpringCloud还是挺机制的，提供回调接口让我们自己实现自己的业务逻辑，真心不错；

  • 那么反过来想想，为啥会无缘无故 start 方法就被调用了呢？那么反向继续向上找调用 start 方法的地方，结果找到了DefaultLifecycleProcessor类的doStart方法调用了 bean.start();

DefaultLifecycleProcessor

EurekaServerInitializerConfiguration.start 方法是如何被触发的？

private void doStart(Map<String, ? extends Lifecycle> lifecycleBeans, String beanName, boolean autoStartupOnly) {
	// 打上断点
	Lifecycle bean = lifecycleBeans.remove(beanName);
	if (bean != null && !this.equals(bean)) {
		String[] dependenciesForBean = this.beanFactory.getDependenciesForBean(beanName);
		for (String dependency : dependenciesForBean) {
			doStart(lifecycleBeans, dependency, autoStartupOnly);
		}
		if (!bean.isRunning() &&
				(!autoStartupOnly || !(bean instanceof SmartLifecycle) || ((SmartLifecycle) bean).isAutoStartup())) {
			if (logger.isDebugEnabled()) {
				logger.debug("Starting bean '" + beanName + "' of type [" + bean.getClass() + "]");
			}
			try {
				bean.start();
			}
			catch (Throwable ex) {
				throw new ApplicationContextException("Failed to start bean '" + beanName + "'", ex);
			}
			if (logger.isDebugEnabled()) {
				logger.debug("Successfully started bean '" + beanName + "'");
			}
		}
	}
}

看到在 bean.isRunning 等一系列状态的判断下才去调用 bean.start() 方法的，我们再往上寻找被调用地方；

public void start() {
	// 打上断点
	if (this.members.isEmpty()) {
		return;
	}
	if (logger.isInfoEnabled()) {
		logger.info("Starting beans in phase " + this.phase);
	}
	Collections.sort(this.members);
	for (LifecycleGroupMember member : this.members) {
		if (this.lifecycleBeans.containsKey(member.name)) {
			doStart(this.lifecycleBeans, member.name, this.autoStartupOnly);
		}
	}
}

该类是DefaultLifecycleProcessor中内部类LifecycleGroup的一个方法，再往上寻找调用方；

private void startBeans(boolean autoStartupOnly) {
	Map<String, Lifecycle> lifecycleBeans = getLifecycleBeans();
	Map<Integer, LifecycleGroup> phases = new HashMap<Integer, LifecycleGroup>();
	for (Map.Entry<String, ? extends Lifecycle> entry : lifecycleBeans.entrySet()) {
		Lifecycle bean = entry.getValue();
		if (!autoStartupOnly || (bean instanceof SmartLifecycle && ((SmartLifecycle) bean).isAutoStartup())) {
			int phase = getPhase(bean);
			LifecycleGroup group = phases.get(phase);
			if (group == null) {
				group = new LifecycleGroup(phase, this.timeoutPerShutdownPhase, lifecycleBeans, autoStartupOnly);
				phases.put(phase, group);
			}
			group.add(entry.getKey(), bean);
		}
	}
	if (phases.size() > 0) {
		List<Integer> keys = new ArrayList<Integer>(phases.keySet());
		Collections.sort(keys);
		for (Integer key : keys) {
			phases.get(key).start();
		}
	}
}

  • startBeans 属于 DefaultLifecycleProcessor 类的一个私有方法，startBeans 方法第一行就是获取 getLifecycleBeans() 生命周期Bean对象，由此可见似乎 Eureka Server 之所以会被启动，是不是实现了某个接口或者重写了某个方法，才会导致由于容易在初始化的过程中因调用某些特殊方法或者某些类才启动的，因此我们回头去看看 EurekaServerInitializerConfiguration 这个类；

  • 结果发现 EurekaServerInitializerConfiguration 这个类实现了 SmartLifecycle 这么样的一个接口，而 SmartLifecycle 接口又继承了 Lifecycle 生命周期接口类，所以真想已经重见天日了，原来是实现了 Lifecycle 这样的一个接口，然后实现了 start 方法，因此 Eureka Server 就这么稀里糊涂的就被莫名其妙的启动起来了？

我们之前仅仅只是通过了日志来逆向分析，但是我们是不是忘了我们本应该标志是Eureka Server的这个注解了呢？没错，我们在分析的过程中
已经将 @EnableEurekaServer 这个注解遗忘了，那么我们现在先回到这个注解类来看看；

EnableEurekaServer

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(EurekaServerConfiguration.class)
public @interface EnableEurekaServer {}

我们不难发现 EnableEurekaServer 类上有个 @Import 注解，引用了一个 class 文件，由此我们进入观察；

EurekaServerConfiguration 类看看，看名称的话，理解的意思大概就是 Eureka Server 配置类；

  • 果不其然，这个类有很多 @Bean、@Configuration 注解过的方法，那是不是我们可以认为刚才 3.1~3.4 的推论是不是就是由于被实例化了这么一个 Bean，然后就慢慢的调用到了 start 方法了呢？

  • 搜索 “Bootstrap” 字样，还真发现了有这么一个方法；

@Bean
public EurekaServerBootstrap eurekaServerBootstrap(PeerAwareInstanceRegistry registry,
		EurekaServerContext serverContext) {
	return new EurekaServerBootstrap(this.applicationInfoManager,
			this.eurekaClientConfig, this.eurekaServerConfig, registry,
			serverContext);
}

  • 既然有这么一个 Bean，那么是不是和刚开始顺着日志逆向分析也是有一定道理的，没有这么一个Bean的存在，那么 DefaultLifecycleProcessor.startBeans 方法中 getLifecycleBeans 的这个也就没那么顺畅被找到了呢？不过我的猜想是这样的，本人没有将源码下载下来，将 eurekaServerBootstrap 方法中的 @Bean 注解注释掉试试，不过推理起来也八九不离十，这个疑问悬念就留给大家尝试尝试吧；

  • 既然找到了一个 @Bean 注解过的方法，那我们再找找其他的一些被注解过的方法，比如一些通用全局用的类似词眼，比如 Context，Bean，Init、Server 之类的；

@Bean
public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs,
		PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) {
	return new DefaultEurekaServerContext(this.eurekaServerConfig, serverCodecs,
			registry, peerEurekaNodes, this.applicationInfoManager);
}

@Bean
public PeerEurekaNodes peerEurekaNodes(PeerAwareInstanceRegistry registry,
		ServerCodecs serverCodecs) {
	return new PeerEurekaNodes(registry, this.eurekaServerConfig,
			this.eurekaClientConfig, serverCodecs, this.applicationInfoManager);
}

@Bean
public PeerAwareInstanceRegistry peerAwareInstanceRegistry(
		ServerCodecs serverCodecs) {
	this.eurekaClient.getApplications(); // force initialization
	return new InstanceRegistry(this.eurekaServerConfig, this.eurekaClientConfig,
			serverCodecs, this.eurekaClient,
			this.instanceRegistryProperties.getExpectedNumberOfRenewsPerMin(),
			this.instanceRegistryProperties.getDefaultOpenForTrafficCount());
}

@Bean
@ConditionalOnProperty(prefix = "eureka.dashboard", name = "enabled", matchIfMissing = true)
public EurekaController eurekaController() {
	return new EurekaController(this.applicationInfoManager);
}

  • DefaultEurekaServerContext.initialize 初始化了一些东西，现在还不知道干啥用的，先放这里，打上断点；

  • PeerEurekaNodes.start 方法，又是一个 start 方法，但是该类没有实现任何类，姑且先放这里，打上断点；

  • InstanceRegistry.register 方法，而且还有几个呢，可能是客户端注册用的，也先放这里，都打上断点，或者将 这个类的所有方法都断点上，断点打完后发现有注册的，有续约的，有注销的；

  • 打完这些断点后，感觉没有思路了，索性就断点跑一把，看看有什么新的发现点；

【分析一】：DefaultEurekaServerContext.initialize 方法被调用了，证实了刚才想法，EurekaServerConfiguration 不是白写的，还是有它的作用的；

@PostConstruct
@Override
public void initialize() throws Exception {
    logger.info("Initializing ...");
    peerEurekaNodes.start();
    registry.init(peerEurekaNodes);
    logger.info("Initialized");
}

【分析二】：进入 initialize 方法中 peerEurekaNodes.start();

public void start() {
    taskExecutor = Executors.newSingleThreadScheduledExecutor(
            new ThreadFactory() {
                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(r, "Eureka-PeerNodesUpdater");
                    thread.setDaemon(true);
                    return thread;
                }
            }
    );
    try {
        updatePeerEurekaNodes(resolvePeerUrls());
        Runnable peersUpdateTask = new Runnable() {
            @Override
            public void run() {
                try {
                    updatePeerEurekaNodes(resolvePeerUrls());
                } catch (Throwable e) {
                    logger.error("Cannot update the replica Nodes", e);
                }

            }
        };
		// 注释：间隔 600000 毫秒，即 10分钟 间隔执行一次服务集群数据同步；
        taskExecutor.scheduleWithFixedDelay(
                peersUpdateTask,
                serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
                serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
                TimeUnit.MILLISECONDS
        );
    } catch (Exception e) {
        throw new IllegalStateException(e);
    }
    for (PeerEurekaNode node : peerEurekaNodes) {
        logger.info("Replica node URL:  " + node.getServiceUrl());
    }
}

  • start 方法中会看到一个定时调度的任务，updatePeerEurekaNodes(resolvePeerUrls()); 间隔 600000 毫秒，即 10分钟 间隔执行一次服务集群数据同步；

  • 然后断点放走放下走，进入 initialize 方法中 registry.init(peerEurekaNodes);

@Override
public void init(PeerEurekaNodes peerEurekaNodes) throws Exception {
    this.numberOfReplicationsLastMin.start();
    this.peerEurekaNodes = peerEurekaNodes;
	// 注释：初始化 Eureka Server 响应缓存，默认缓存时间为30s
    initializedResponseCache();
	// 注释：定时任务，多久重置一下心跳阈值，900000 毫秒，即 15分钟 的间隔时间，会重置心跳阈值
    scheduleRenewalThresholdUpdateTask();
	// 注释：初始化远端注册
    initRemoteRegionRegistry();

    try {
        Monitors.registerObject(this);
    } catch (Throwable e) {
        logger.warn("Cannot register the JMX monitor for the InstanceRegistry :", e);
    }
}

缓存也配置好了，定时任务也配置好了，似乎应该没啥了，那么我们把断点放开，看看下一步会走到哪里？

EurekaServerInitializerConfiguration.start。

  • 先是 DefaultLifecycleProcessor.doStart 方法进断点，然后才是 EurekaServerInitializerConfiguration.start 方法进断点；

  • 再一次证明刚刚的逆向分析仅仅只是缺了个从头EnableEurekaServer分析罢了，但是最终方法论分析思路还是对的，由于开始分析过这里，然而我们就跳过，继续放开断点向后继续看看；

InstanceRegistry.openForTraffic

  • 【这不就是我们刚才在 “步骤3.7之分析七” 打的断点么？看下堆栈信息，正是 “步骤3.2之分析一” 中 initEurekaServerContext 方法中有这么一句 this.registry.openForTraffic(this.applicationInfoManager, registryCount); 调用到了，因果轮回，代码千变万化，打上断点还有有好处的，结果还是回到了开始日志逆向分析的地方。

c进入 super.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.
	// 注释：每30秒续约一次，那么每分钟续约就是2次，所以才是 count * 2 的结果；
    this.expectedNumberOfRenewsPerMin = count * 2;
    this.numberOfRenewsPerMinThreshold =
            (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());
    logger.info("Got " + count + " instances from neighboring DS node");
    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");
	// 注释：修改 Eureka Server 为上电状态，就是说设置 Eureka Server 已经处于活跃状态了，那就是意味着 EurekaServer 基本上说可以正常使用了；
    applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
	// 注释：定时任务，60000 毫秒，即 1分钟 的间隔时间，Eureke Server定期进行失效节点的清理
    super.postInit();
}
  • 这里主要设置了服务状态，以及开启了定时清理失效节点的定时任务，每分钟扫描一次；

继续放开断点，来到了日志打印 “main] c.n.e.EurekaDiscoveryClientConfiguration : Updating port to 8761” 的EurekaDiscoveryClientConfiguration 类中 onApplicationEvent 方法。

@EventListener(EmbeddedServletContainerInitializedEvent.class)
public void onApplicationEvent(EmbeddedServletContainerInitializedEvent event) {
	// TODO: take SSL into account when Spring Boot 1.2 is available
	int localPort = event.getEmbeddedServletContainer().getPort();
	if (this.port.get() == 0) {
		log.info("Updating port to " + localPort);
		this.port.compareAndSet(0, localPort);
		start();
	}
}
  • 设置端口，当看到 Updating port to 8761 这样的日志打印出来的话，说明 Eureka Server 整个启动也就差不多Over了。现在回头看看，发现分析了不少的方法和流程，有种感觉被掏空的感觉了。

总结 EurekaServer 启动时候大概干了哪些事情？

1、初始化Eureka环境，Eureka上下文；
2、初始化EurekaServer的缓存
3、启动了一些定时任务，比如充值心跳阈值定时任务，清理失效节点定时任务；
4、更新EurekaServer上电状态，更新EurekaServer端口；

版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 举报，一经查实，本站将立刻删除。

文章由极客之音整理，本文链接：https://www.bmabk.com/index.php/post/70143.html

赞 (0)
小半的头像小半
0 0

相关推荐

极客之音——专业性很强的中文编程技术网站，欢迎收藏到浏览器，订阅我们！