Eureka 源碼分析之 Eureka Server

文章首發(fā)于公眾號《程序員果果》
地址 : https://mp.weixin.qq.com/s/FfJrAGQuHyVrsedtbr0Ihw

簡介

上一篇文章《Eureka 源碼分析之 Eureka Client》 通過源碼知道 ，eureka Client 是通過 http rest來 與 eureka server 交互，實現(xiàn) 注冊服務，續(xù)約服務，服務下線 等。本篇探究下eureka server。

源碼分析

從 @EnableEurekaServer 注解為入口分析，通過源碼可以看出他是一個標記注解：

/**
 * Annotation to activate Eureka Server related configuration {@link 
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Import(EurekaServerMarkerConfiguration.class)
public @interface EnableEurekaServer {

}

從注釋可以知道，用來激活 eureka server 的 配置類 EurekaServerAutoConfiguration 中相關配置，EurekaServerAutoConfiguration 的關鍵代碼如下：

@Configuration
@Import(EurekaServerInitializerConfiguration.class)
@ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class)
@EnableConfigurationProperties({ EurekaDashboardProperties.class,
        InstanceRegistryProperties.class })
@PropertySource("classpath:/eureka/server.properties")
public class EurekaServerAutoConfiguration extends WebMvcConfigurerAdapter {
    /**
     * List of packages containing Jersey resources required by the Eureka server
     */
    private static final String[] EUREKA_PACKAGES = new String[] { "com.netflix.discovery",
            "com.netflix.eureka" };

    @Autowired
    private ApplicationInfoManager applicationInfoManager;

    @Autowired
    private EurekaServerConfig eurekaServerConfig;

    @Autowired
    private EurekaClientConfig eurekaClientConfig;

    @Autowired
    private EurekaClient eurekaClient;

    @Autowired
    private InstanceRegistryProperties instanceRegistryProperties;

    public static final CloudJacksonJson JACKSON_JSON = new CloudJacksonJson();

    @Bean
    public HasFeatures eurekaServerFeature() {
        return HasFeatures.namedFeature("Eureka Server",
                EurekaServerAutoConfiguration.class);
    }

    @Configuration
    protected static class EurekaServerConfigBeanConfiguration {
        // 創(chuàng)建并加載EurekaServerConfig的實現(xiàn)類，主要是Eureka-server的配置信息
        @Bean
        @ConditionalOnMissingBean
        public EurekaServerConfig eurekaServerConfig(EurekaClientConfig clientConfig) {
            EurekaServerConfigBean server = new EurekaServerConfigBean();
            if (clientConfig.shouldRegisterWithEureka()) {
                // Set a sensible default if we are supposed to replicate
                server.setRegistrySyncRetries(5);
            }
            return server;
        }
    }

    //加載EurekaController，SpringCloud 提供了一些額外的接口，用來獲取eurekaServer的信息
    @Bean
    @ConditionalOnProperty(prefix = "eureka.dashboard", name = "enabled", matchIfMissing = true)
    public EurekaController eurekaController() {
        return new EurekaController(this.applicationInfoManager);
    }

    //省略 ...

    // 接收客戶端的注冊等請求就是通過InstanceRegistry來處理的，是真正處理業(yè)務的類，接下來會詳細分析
    @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());
    }

    //配置服務節(jié)點信息，這里的作用主要是為了配置Eureka的peer節(jié)點，也就是說當有收到有節(jié)點注冊上來的時候，需要通知給哪些節(jié)點
    @Bean
    @ConditionalOnMissingBean
    public PeerEurekaNodes peerEurekaNodes(PeerAwareInstanceRegistry registry,
            ServerCodecs serverCodecs) {
        return new RefreshablePeerEurekaNodes(registry, this.eurekaServerConfig,
                this.eurekaClientConfig, serverCodecs, this.applicationInfoManager);
    }

    //省略 ...    

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

    // 初始化Eureka-server，會同步其他注冊中心的數(shù)據(jù)到當前注冊中心
    @Bean
    public EurekaServerBootstrap eurekaServerBootstrap(PeerAwareInstanceRegistry registry,
            EurekaServerContext serverContext) {
        return new EurekaServerBootstrap(this.applicationInfoManager,
                this.eurekaClientConfig, this.eurekaServerConfig, registry,
                serverContext);
    }

    // 配置攔截器，ServletContainer里面實現(xiàn)了jersey框架，通過他來實現(xiàn)eurekaServer對外的restFull接口
    @Bean
    public FilterRegistrationBean jerseyFilterRegistration(
            javax.ws.rs.core.Application eurekaJerseyApp) {
        FilterRegistrationBean bean = new FilterRegistrationBean();
        bean.setFilter(new ServletContainer(eurekaJerseyApp));
        bean.setOrder(Ordered.LOWEST_PRECEDENCE);
        bean.setUrlPatterns(
                Collections.singletonList(EurekaConstants.DEFAULT_PREFIX + "/*"));

        return bean;
    }

    //省略 ...

}

從EurekaServerAutoConfiguration 類上的注解@Import(EurekaServerInitializerConfiguration.class) 可以到，實例化類EurekaServerAutoConfiguration之前，已經(jīng)實例化了EurekaServerInitializerConfiguration類，代碼如下：

@Configuration
@CommonsLog
public class EurekaServerInitializerConfiguration
      implements ServletContextAware, SmartLifecycle, Ordered {

   // 此處省略部分代碼

   @Override
   public void start() {
      // 啟動一個線程
      new Thread(new Runnable() {
         @Override
         public void run() {
            try {
               //初始化EurekaServer，同時啟動Eureka Server ，后面著重講這里
               eurekaServerBootstrap.contextInitialized(EurekaServerInitializerConfiguration.this.servletContext);
               log.info("Started Eureka Server");
                // 發(fā)布EurekaServer的注冊事件
               publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));
                // 設置啟動的狀態(tài)為true
               EurekaServerInitializerConfiguration.this.running = true;
                // 發(fā)送Eureka Start 事件 ， 其他還有各種事件，我們可以監(jiān)聽這種時間，然后做一些特定的業(yè)務需求，后面會講到。
               publish(new EurekaServerStartedEvent(getEurekaServerConfig()));
            }
            catch (Exception ex) {
               // Help!
               log.error("Could not initialize Eureka servlet context", ex);
            }
         }
      }).start();
   }

   // 此處省略部分代碼

}

這個start方法中開啟了一個新的線程，然后進行一些Eureka Server的初始化工作，比如調用eurekaServerBootstrap的contextInitialized方法，EurekaServerBootstrap代碼如下：

public class EurekaServerBootstrap {

    // 此處省略部分代碼

    public void contextInitialized(ServletContext context) {
       try {
          // 初始化Eureka的環(huán)境變量
          initEurekaEnvironment();
          // 初始化Eureka的上下文
          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);
       }
    }

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

       String dataCenter = ConfigurationManager.getConfigInstance()
             .getString(EUREKA_DATACENTER);
       if (dataCenter == null) {
          log.info(
                "Eureka data center value eureka.datacenter is not set, defaulting to default");
          ConfigurationManager.getConfigInstance()
                .setProperty(ARCHAIUS_DEPLOYMENT_DATACENTER, DEFAULT);
       }
       else {
          ConfigurationManager.getConfigInstance()
                .setProperty(ARCHAIUS_DEPLOYMENT_DATACENTER, dataCenter);
       }
       String environment = ConfigurationManager.getConfigInstance()
             .getString(EUREKA_ENVIRONMENT);
       if (environment == null) {
          ConfigurationManager.getConfigInstance()
                .setProperty(ARCHAIUS_DEPLOYMENT_ENVIRONMENT, TEST);
          log.info(
                "Eureka environment value eureka.environment is not set, defaulting to test");
       }
       else {
          ConfigurationManager.getConfigInstance()
                .setProperty(ARCHAIUS_DEPLOYMENT_ENVIRONMENT, environment);
       }
    }

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

       //初始化eureka server上下文
       EurekaServerContextHolder.initialize(this.serverContext);

       log.info("Initialized server context");

       // Copy registry from neighboring eureka node
       // 從相鄰的eureka節(jié)點復制注冊表 
       int registryCount = this.registry.syncUp();
        // 默認每30秒發(fā)送心跳，1分鐘就是2次
        // 修改eureka狀態(tài)為up 
        // 同時，這里面會開啟一個定時任務，用于清理 60秒沒有心跳的客戶端。自動下線
       this.registry.openForTraffic(this.applicationInfoManager, registryCount);

       // Register all monitoring statistics.
       EurekaMonitors.registerAllStats();
    }
    public void contextDestroyed(ServletContext context) {
       try {
          log.info("Shutting down Eureka Server..");
          context.removeAttribute(EurekaServerContext.class.getName());

          destroyEurekaServerContext();
          destroyEurekaEnvironment();

       }
       catch (Throwable e) {
          log.error("Error shutting down eureka", e);
       }
       log.info("Eureka Service is now shutdown...");
    }

}

在初始化Eureka Server上下文環(huán)境后，就會繼續(xù)執(zhí)行openForTraffic方法，這個方法主要是設置了期望每分鐘接收到的心跳次數(shù)，并將服務實例的狀態(tài)設置為UP，最后又通過方法postInit來開啟一個定時任務，用于每隔一段時間（默認60秒）將沒有續(xù)約的服務實例（默認90秒沒有續(xù)約）清理掉。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.
    // 計算每分鐘最大續(xù)約數(shù)
    this.expectedNumberOfRenewsPerMin = count * 2;
    // 計算每分鐘最小續(xù)約數(shù)
    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");
    // 修改服務實例的狀態(tài)為UP
    applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
    // 開啟定時任務，每隔一段時間（默認60秒）將沒有續(xù)約的服務實例（默認90秒沒有續(xù)約）清理掉
    super.postInit();
}

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

這里的時間間隔都來自于EurekaServerConfigBean類，可以在配置文件中以eureka.server開頭的配置來進行設置。

參考

https://www.e-learn.cn/content/qita/775244/
https://nobodyiam.com/2016/06/25/dive-into-eureka/
https://blog.csdn.net/Lammonpeter/article/details/84330900

關注我