Eureka源码深度解析（3）

系统管理员 2022-02-01 04:47 289阅读 0赞

### Eureka Server启动过程 ###

同Eureka Client启动一样，需要添加[@EnableEurekaServer][EnableEurekaServer]注解。在该类中用[@Import][Import](EurekaServerMarkerConfiguration.class)表明了程序在启动时会先加载EurekaServerMarkerConfiguration配置类中的配置，而在该配置类中，发布了一个标记类 EurekaServerMarkerConfiguration$Marker，该标记类会用于开启后续EurekaServer相关配置的加载工作。

org.springframework.cloud.netflix.eureka.server.EnableEurekaServer

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

org.springframework.cloud.netflix.eureka.server.EurekaServerMarkerConfiguration

@Configuration
    public class EurekaServerMarkerConfiguration {
        public EurekaServerMarkerConfiguration() {
        }
    
        @Bean
        public EurekaServerMarkerConfiguration.Marker eurekaServerMarkerBean() {
            return new EurekaServerMarkerConfiguration.Marker();
        }
    
        class Marker {
            Marker() {
            }
        }
    }

我们看到只是实例化了一个空类，没有任何实现，从注释中可以看到 EurekaServerMarkerConfiguration 是一个激活 EurekaServerAutoConfiguration 的开关。通过之前的分析，我们实际可以发现SpringBoot相关项目的一些设计模式了，很多的类并不是被显示的加载到容器中，而是通过配置的方式，最经典的方式就是放到META-INF/spring.factories文件中去加载，那么我们也来看下spring-cloud-netflix-eureka-server-\{version\}.jar!META-INF/spring.factories，具体内容如下：

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
      org.springframework.cloud.netflix.eureka.server.EurekaServerAutoConfiguration

真正的配置信息在 EurekaServerAutoConfiguration 中，我们看到 @ConditionalOnBean(Marker.class) 只有存在 Marker 实例的时候，才会继续加载配置项，这也就要求必须有 @EnableEurekaServer 注解，才能正常的启动。

源码如下：

@Configuration  //表明这是一个配置类
    @Import({EurekaServerInitializerConfiguration.class})  //导入启动EurekaServer的bean
    @ConditionalOnBean({Marker.class})  //这个是表示只有在spring容器里面含有Market这个实例的时候，才会加载当前这个Bean（EurekaServerAutoConfiguration ），这个就是控制是否开启EurekaServer的关键
    @EnableConfigurationProperties({EurekaDashboardProperties.class, InstanceRegistryProperties.class})
    @PropertySource({"classpath:/eureka/server.properties"})   //加载配置文件
    public class EurekaServerAutoConfiguration extends WebMvcConfigurerAdapter {
    
        ...
    
        //Eureka-server的可视化界面就是通过EurekaController提供的
        public EurekaController eurekaController() {
            return new EurekaController(this.applicationInfoManager);
        }
        ...
        //接收客户端的注册等请求就是通过InstanceRegistry来处理的，是真正处理业务的类
        @Bean
        public PeerAwareInstanceRegistry peerAwareInstanceRegistry(ServerCodecs serverCodecs) {
            this.eurekaClient.getApplications();
            return new InstanceRegistry(this.eurekaServerConfig, this.eurekaClientConfig, serverCodecs, this.eurekaClient, this.instanceRegistryProperties.getExpectedNumberOfRenewsPerMin(), this.instanceRegistryProperties.getDefaultOpenForTrafficCount());
        }
    
        ...
    
        //初始化Eureka-server，会同步其他注册中心的数据到当前注册中心
        @Bean
        public EurekaServerBootstrap eurekaServerBootstrap(PeerAwareInstanceRegistry registry, EurekaServerContext serverContext) {
            return new EurekaServerBootstrap(this.applicationInfoManager, this.eurekaClientConfig, this.eurekaServerConfig, registry, serverContext);
        }
    
        @Bean
        public FilterRegistrationBean jerseyFilterRegistration(Application eurekaJerseyApp) {
            FilterRegistrationBean bean = new FilterRegistrationBean();
            bean.setFilter(new ServletContainer(eurekaJerseyApp));
            bean.setOrder(2147483647);
    	//创建Filter,并匹配路径/eureka/*
            bean.setUrlPatterns(Collections.singletonList("/eureka/*"));
            return bean;
        }
    
        @Bean
        public Application jerseyApplication(Environment environment, ResourceLoader resourceLoader) {
            ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(false, environment);
            //创建相关的web节点， 比如注册接口/eureka/apps/{appId}
    	provider.addIncludeFilter(new AnnotationTypeFilter(Path.class));
            provider.addIncludeFilter(new AnnotationTypeFilter(Provider.class));
            Set<Class<?>> classes = new HashSet();
            String[] var5 = EUREKA_PACKAGES;
            int var6 = var5.length;
    	//扫描restful接口资源的类
            for(int var7 = 0; var7 < var6; ++var7) {
                String basePackage = var5[var7];
                Set<BeanDefinition> beans = provider.findCandidateComponents(basePackage);
                Iterator var10 = beans.iterator();
    
                while(var10.hasNext()) {
                    BeanDefinition bd = (BeanDefinition)var10.next();
                    Class<?> cls = ClassUtils.resolveClassName(bd.getBeanClassName(), resourceLoader.getClassLoader());
                    classes.add(cls);
                }
            }
    
            Map<String, Object> propsAndFeatures = new HashMap();
            propsAndFeatures.put("com.sun.jersey.config.property.WebPageContentRegex", "/eureka/(fonts|images|css|js)/.*");
            DefaultResourceConfig rc = new DefaultResourceConfig(classes);
            rc.setPropertiesAndFeatures(propsAndFeatures);
            return rc;
        }
    
        ...
    
        //创建并加载EurekaServerConfig的实现类，主要是Eureka-server的配置信息
        @Configuration
        protected static class EurekaServerConfigBeanConfiguration {
            protected EurekaServerConfigBeanConfiguration() {
            }
    
            @Bean
            @ConditionalOnMissingBean
            public EurekaServerConfig eurekaServerConfig(EurekaClientConfig clientConfig) {
                EurekaServerConfigBean server = new EurekaServerConfigBean();
                if (clientConfig.shouldRegisterWithEureka()) {
    		//当eureka服务器启动时尝试去获取集群里其他服务器上的注册信息的次数，默认为5
                    server.setRegistrySyncRetries(5);
                }
    
                return server;
            }
        }
    }

通过以上分析可知，EurekaServer在启动的时候，会加载很多bean到Spring容器中，每个bean都实现了各自的功能，其中真正处理客户端请求的类是 InstanceRegistry

package org.springframework.cloud.netflix.eureka.server;
    
    public class InstanceRegistry extends PeerAwareInstanceRegistryImpl implements ApplicationContextAware {
    
        private static final Log log = LogFactory.getLog(InstanceRegistry.class);
        private ApplicationContext ctxt;
        private int defaultOpenForTrafficCount;
        ...
        // 1.接收客户端注册请求
        public void register(InstanceInfo info, int leaseDuration, boolean isReplication) {
            this.handleRegistration(info, leaseDuration, isReplication);
            super.register(info, leaseDuration, isReplication);
        }
     
        // 2.接收客户端下线请求
        public boolean cancel(String appName, String serverId, boolean isReplication) {
            this.handleCancelation(appName, serverId, isReplication);
            return super.cancel(appName, serverId, isReplication);
        }
     
        // 3.接收客户端续约请求
        public boolean renew(final String appName, final String serverId, boolean isReplication) {
            this.log("renew " + appName + " serverId " + serverId + ", isReplication {}" + isReplication);
            List<Application> applications = this.getSortedApplications();
            Iterator var5 = applications.iterator();
    
            while(var5.hasNext()) {
                Application input = (Application)var5.next();
                if (input.getName().equals(appName)) {
                    InstanceInfo instance = null;
                    Iterator var8 = input.getInstances().iterator();
    
                    while(var8.hasNext()) {
                        InstanceInfo info = (InstanceInfo)var8.next();
                        if (info.getId().equals(serverId)) {
                            instance = info;
                            break;
                        }
                    }
    
                    this.publishEvent(new EurekaInstanceRenewedEvent(this, appName, serverId, instance, isReplication));
                    break;
                }
            }
    
            return super.renew(appName, serverId, isReplication);
        }
    
        ...
    
    }

以上是在处理客户端的不同请求，但是，客户端发送的是HTTP请求，这只是一个类，服务端应该也有一个接收HTTP请求的类，然后将接收到的请求封装后委托给InstanceRegistry来处理具体业务。这个类就是com.netflix.eureka.resources包下的ApplicationResource、InstanceResource。

**Register原理（服务注册）**

这个接口会在Service Provider启动时被调用来实现服务注册。同时，当Service Provider的服务状态发生变化时（如自身检测认为Down的时候），也会调用来更新服务状态。接口实现比较简单，如下图所示。

1.  `ApplicationResource`类接收Http服务请求，调用`PeerAwareInstanceRegistryImpl`的`register`方法
2.  `PeerAwareInstanceRegistryImpl`完成服务注册后，调用`replicateToPeers`向其它Eureka Server节点（Peer）做状态同步

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70][]

源码：AbstractInstanceRegistry.register(InstanceInfo registrant, int leaseDuration, boolean isReplication)注册

public void register(InstanceInfo registrant, int leaseDuration, boolean isReplication) {
            try {
                this.read.lock();
                // 所有的服务信息都添加到registry这个map中，
                // 格式为：ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>()
                Map<String, Lease<InstanceInfo>> gMap = registry.get(registrant.getAppName());
                REGISTER.increment(isReplication);
                // 如果没有该服务的信息，则新建，并添加到registry中
                if (gMap == null) {
                    final ConcurrentHashMap<String, Lease<InstanceInfo>> gNewMap = new ConcurrentHashMap<String, Lease<InstanceInfo>>();
                    gMap = registry.putIfAbsent(registrant.getAppName(), gNewMap);
                    if (gMap == null) {
                        gMap = gNewMap;
                    }
                }
                //existingLease信息即服务的一些注册时间等信息，主要是为了校验该服务是否过期，如果已过期，则剔除
                Lease<InstanceInfo> existingLease = gMap.get(registrant.getId());
             
                if (existingLease != null && (existingLease.getHolder() != null)) {
                    Long existingLastDirtyTimestamp = existingLease.getHolder().getLastDirtyTimestamp();
                    Long registrationLastDirtyTimestamp = registrant.getLastDirtyTimestamp();
                    logger.debug("Existing lease found (existing={}, provided={}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
                    if (existingLastDirtyTimestamp > registrationLastDirtyTimestamp) {
                        logger.warn("There is an existing lease and the existing lease's dirty timestamp {} is greater" +
                                " than the one that is being registered {}", existingLastDirtyTimestamp, registrationLastDirtyTimestamp);
                        logger.warn("Using the existing instanceInfo instead of the new instanceInfo as the registrant");
                        registrant = existingLease.getHolder();
                    }
                } else {
                    synchronized (lock) {
                        if (this.expectedNumberOfRenewsPerMin > 0) {
                            this.expectedNumberOfRenewsPerMin = this.expectedNumberOfRenewsPerMin + 2;
                            this.numberOfRenewsPerMinThreshold =
                                    (int) (this.expectedNumberOfRenewsPerMin * serverConfig.getRenewalPercentThreshold());
                        }
                    }
                    logger.debug("No previous lease information found; it is new registration");
                }
                Lease<InstanceInfo> lease = new Lease<InstanceInfo>(registrant, leaseDuration);
                if (existingLease != null) {
                    lease.setServiceUpTimestamp(existingLease.getServiceUpTimestamp());
                }
                gMap.put(registrant.getId(), lease);
                ...
            } finally {
                read.unlock();
            }
        }

服务注册信息最终存放到 ConcurrentHashMap<String, Map<String, Lease<InstanceInfo>>>，外层map的key即为应用的服务名，内层map的key为我们设置的eureka.instance.instance-id，设置成这种格式，当多个应用提供相同服务时，那么外层map的key都相同，内层map的key不同。

**Renew（服务续约）**

由Service Provider定期调用，类似于heartbeat。主要是用来告诉Eureka Server Service Provider还活着，避免服务被剔除掉。接口实现如下图所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70 1][]

可以看到，接口实现方式和`register`基本一致：首先更新自身状态，再同步到其它Peer。

源码：AbstractInstanceRegistry.renew(String appName, String id, boolean isReplication)续约

public boolean renew(String appName, String id, boolean isReplication) {
            EurekaMonitors.RENEW.increment(isReplication);
            // 1.获取对应map
            Map<String, Lease<InstanceInfo>> gMap = registry.get(appName);
            Lease<InstanceInfo> leaseToRenew = null;
            if (gMap != null) {
                // 2.主要是为了获取当前服务的一些过期信息
                leaseToRenew = gMap.get(id);
            }
            ...
                renewsLastMin.increment();
                // 主要操作在这里，将最新更新时间重置，剔除任务检查的也就是这个最新更新时间
                // lastUpdateTimestamp = System.currentTimeMillis() + duration;
                leaseToRenew.renew();
                return true;
            }
        }

**Cancel（服务下线）**

一般在Service Provider shut down的时候调用，用来把自身的服务从Eureka Server中删除，以防客户端调用不存在的服务。接口实现如下图所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70 2][]

源码：AbstractInstanceRegistry.cancel(String appName, String id, boolean isReplication)下线

protected boolean internalCancel(String appName, String id, boolean isReplication) {
            boolean var7;
            try {
                this.read.lock();
                EurekaMonitors.CANCEL.increment(isReplication);
    	    // 1.获取gmap
                Map<String, Lease<InstanceInfo>> gMap = (Map)this.registry.get(appName);
                Lease<InstanceInfo> leaseToCancel = null;
                if (gMap != null) {
    		// 2.删除gmap中该服务id
                    leaseToCancel = (Lease)gMap.remove(id);
                }
    
                AbstractInstanceRegistry.CircularQueue var6 = this.recentCanceledQueue;
                synchronized(this.recentCanceledQueue) {
                    this.recentCanceledQueue.add(new Pair(System.currentTimeMillis(), appName + "(" + id + ")"));
                }
    
    	    
                InstanceStatus instanceStatus = (InstanceStatus)this.overriddenInstanceStatusMap.remove(id);
                if (instanceStatus != null) {
                    logger.debug("Removed instance id {} from the overridden map which has value {}", id, instanceStatus.name());
                }
    
                if (leaseToCancel != null) {
    		//3.将当前服务的剔除时间置为当前时间 evictionTimestamp = System.currentTimeMillis();
                    leaseToCancel.cancel();
    		// 4.获取服务信息
                    InstanceInfo instanceInfo = (InstanceInfo)leaseToCancel.getHolder();
                    String vip = null;
                    String svip = null;
                    if (instanceInfo != null) {
    		    // 5.将服务信息置为已删除
                        instanceInfo.setActionType(ActionType.DELETED);
                        this.recentlyChangedQueue.add(new AbstractInstanceRegistry.RecentlyChangedItem(leaseToCancel));
                        instanceInfo.setLastUpdatedTimestamp();
                        vip = instanceInfo.getVIPAddress();
                        svip = instanceInfo.getSecureVipAddress();
                    }
    
                    this.invalidateCache(appName, vip, svip);
                    logger.info("Cancelled instance {}/{} (replication={})", new Object[]{appName, id, isReplication});
                    boolean var10 = true;
                    return var10;
                }
    
                EurekaMonitors.CANCEL_NOT_FOUND.increment(isReplication);
                logger.warn("DS: Registry: cancel failed because Lease is not registered for: {}/{}", appName, id);
                var7 = false;
            } finally {
                this.read.unlock();
            }
    
            return var7;
        }

综上所述，简单来讲，服务的注册实际上是将服务信息添加到一个map中，map的key是服务名称，value也是一个map，是提供该服务的所有客户端信息； 服务的续约实际上是获取map中该服务的客户端信息，然后修改其最新更新时间；服务的下线实际上是删除该map中该服务信息，然后修改服务状态。

[EnableEurekaServer]: https://github.com/EnableEurekaServer
[Import]: https://github.com/Import
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70]: /images/20220201/6e82e0809ee64f5da9f9e3730eb1b3d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70 1]: /images/20220201/3253824071924a659c8d59515e4b4c4b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2plcnJ5RHphbg_size_16_color_FFFFFF_t_70 2]: /images/20220201/612afa6d2002447ab8e2b6b10275a3d5.png