SpringCloud-技术专区-从源码层面让你认识Eureka工作流程和运作机制（上）

Posted 2021-09-11 李浩宇Alex

前言介绍

基本原理

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微服务已启动.

EurekaServerBootstrap类

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

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

EurekaServerInitializerConfiguration

进入 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

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 + "\'");
            }
        }
    }
}

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

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 就这么稀里糊涂的就被莫名其妙的启动起来了？

EnableEurekaServer

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

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); 调用到了，因果轮回，代码千变万化，打上断点还有有好处的，结果还是回到了开始日志逆向分析的地方。

@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();
}

• 这里主要设置了服务状态，以及开启了定时清理失效节点的定时任务，每分钟扫描一次；

@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端口；