美文网首页
Eureka 源码分析之 Eureka Server

Eureka 源码分析之 Eureka Server

作者: 程序员果果 | 来源:发表于2019-06-13 10:19 被阅读0次

    文章首发于公众号《程序员果果》
    地址 : https://mp.weixin.qq.com/s/FfJrAGQuHyVrsedtbr0Ihw

    简介

    上一篇文章《Eureka 源码分析之 Eureka Client》 通过源码知道 ,eureka Client 是通过 http rest来 与 eureka server 交互,实现 注册服务,续约服务,服务下线 等。本篇探究下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 {
            // 创建并加载EurekaServerConfig的实现类,主要是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来处理的,是真正处理业务的类,接下来会详细分析
        @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());
        }
        
        //配置服务节点信息,这里的作用主要是为了配置Eureka的peer节点,也就是说当有收到有节点注册上来的时候,需要通知给哪些节点
        @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,会同步其他注册中心的数据到当前注册中心
        @Bean
        public EurekaServerBootstrap eurekaServerBootstrap(PeerAwareInstanceRegistry registry,
                EurekaServerContext serverContext) {
            return new EurekaServerBootstrap(this.applicationInfoManager,
                    this.eurekaClientConfig, this.eurekaServerConfig, registry,
                    serverContext);
        }
    
        // 配置拦截器,ServletContainer里面实现了jersey框架,通过他来实现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之前,已经实例化了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");
                    // 发布EurekaServer的注册事件
                   publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));
                    // 设置启动的状态为true
                   EurekaServerInitializerConfiguration.this.running = true;
                    // 发送Eureka Start 事件 , 其他还有各种事件,我们可以监听这种时间,然后做一些特定的业务需求,后面会讲到。
                   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的环境变量
              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节点复制注册表 
           int registryCount = this.registry.syncUp();
            // 默认每30秒发送心跳,1分钟就是2次
            // 修改eureka状态为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上下文环境后,就会继续执行openForTraffic方法,这个方法主要是设置了期望每分钟接收到的心跳次数,并将服务实例的状态设置为UP,最后又通过方法postInit来开启一个定时任务,用于每隔一段时间(默认60秒)将没有续约的服务实例(默认90秒没有续约)清理掉。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");
        // 修改服务实例的状态为UP
        applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
        // 开启定时任务,每隔一段时间(默认60秒)将没有续约的服务实例(默认90秒没有续约)清理掉
        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

    关注我

    相关文章

      网友评论

          本文标题:Eureka 源码分析之 Eureka Server

          本文链接:https://www.haomeiwen.com/subject/loljfctx.html