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

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)
复制代码

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

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);
  }
}
复制代码

@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();
}
复制代码

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();
    }
  }
}
复制代码

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

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

@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 方法被调用了，证实了刚才想法，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());
    }
}
复制代码

@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);
    }
}
复制代码

@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();
  }
}
复制代码