Eureka源码阅读

ゞ浴缸里的玫瑰 2022-12-26 04:53 256阅读 0赞

# Eureka源码阅读 #

## Eureka中的一些概念 ##

在阅读Eureka源码之前我们需要弄清楚几个概念:

*  Register: 服务注册
 *  Renew: 服务续约
 *  Fetch Registries: 获取注册列表信息
 *  Cancel: 服务下线
 *  Eviction: 服务提出

## Eureka的高可用架构 ##

如图为Eureka的高级架构图，该图片来自于Eureka开源代码的[文档][Link 1]

![e87af352e5a889951fa8d368c7d46fd4.png][]

从图可以看出在这个体系中，有2个角色，即Eureka Server和Eureka Client。而Eureka Client又分为Applicaton Service和Application Client，即服务提供者何服务消费者。 每个区域有一个Eureka集群，并且每个区域至少有一个eureka服务器可以处理区域故障，以防服务器瘫痪。  
Eureka Client向Eureka Serve注册，并将自己的一些客户端信息发送Eureka Serve。然后，Eureka Client通过向Eureka Serve发送心跳（每30秒）来续约服务的。 如果客户端持续不能续约，那么，它将在大约90秒内从服务器注册表中删除。 注册信息和续订被复制到集群中的Eureka Serve所有节点。 来自任何区域的Eureka Client都可以查找注册表信息（每30秒发生一次）。根据这些注册表信息，Application Client可以远程调用Applicaton Service来消费服务。

## Eureka-Client ##

在解读Eureka-Client之前我们先来找一找代码的入口在哪里.  
对于SpringBoot的项目而言就算说一切都给予start都不为过, 我们在maven中找到Eureka-Client的start.

下面这张图是我梳理之后组合的一个类图, 从下往上可以看到在初始化加载EurekaClient时start帮我们做了哪些事情.

1.  读取yaml文件以及机器信息获取一些默认的值;
2.  将读取到的结果封装到EurekaInstanceConfigBean和EurekaClientConfigBean中;
3.  组合需要的参数创建CloudEurekaClient;
4.  完成EurekaClient创建.

![3ee0caeec9aaa716f3dcfdc805dbd38e.png][]

在SpringBoot完成包装之后我们来看一下实际创建的细节, 这部分的逻辑属于Netflix的Eureka.

### DiscoverClient ###

@Singleton
    public class DiscoveryClient implements EurekaClient { 
        @Inject
        DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
                        Provider<BackupRegistry> backupRegistryProvider, EndpointRandomizer endpointRandomizer) { 
            scheduler = Executors.newScheduledThreadPool(2,
                    new ThreadFactoryBuilder()
                            .setNameFormat("DiscoveryClient-%d")
                            .setDaemon(true)
                            .build());
    
            heartbeatExecutor = new ThreadPoolExecutor(
                    1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>(),
                    new ThreadFactoryBuilder()
                            .setNameFormat("DiscoveryClient-HeartbeatExecutor-%d")
                            .setDaemon(true)
                            .build()
            );  // use direct handoff
    
            cacheRefreshExecutor = new ThreadPoolExecutor(
                    1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
                    new SynchronousQueue<Runnable>(),
                    new ThreadFactoryBuilder()
                            .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
                            .setDaemon(true)
                            .build()
            );  // use direct handoff
            // finally, init the schedule tasks 
            initScheduledTasks();
        }
        private void initScheduledTasks() { 
            if (clientConfig.shouldFetchRegistry()) { 
                // registry cache refresh timer
                ...
            }
            if (clientConfig.shouldRegisterWithEureka()) { 
                // Heartbeat timer
            }
            // registry statue change listener
            statusChangeListener = new ApplicationInfoManager.StatusChangeListener() { ...}
            if (clientConfig.shouldOnDemandUpdateStatusChange()) { 
                applicationInfoManager.registerStatusChangeListener(statusChangeListener);
            }
        }
    
        // Register with the eureka service by making the appropriate REST call.
        boolean register() throws Throwable { 
            EurekaHttpResponse<Void> httpResponse;
            httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
            return httpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode();
        }
    
        // Renew with the eureka service by making the appropriate REST call
        boolean renew() { 
            EurekaHttpResponse<InstanceInfo> httpResponse;
            httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
            logger.debug(PREFIX + "{} - Heartbeat status: {}", appPathIdentifier, httpResponse.getStatusCode());
            // check response if 404 will register again
            if (httpResponse.getStatusCode() == Status.NOT_FOUND.getStatusCode()) { 
                REREGISTER_COUNTER.increment();
                logger.info(PREFIX + "{} - Re-registering apps/{}", appPathIdentifier, instanceInfo.getAppName());
                long timestamp = instanceInfo.setIsDirtyWithTime();
                boolean success = register();
                if (success) { 
                    instanceInfo.unsetIsDirty(timestamp);
                }
                return success;
            }
            return httpResponse.getStatusCode() == Status.OK.getStatusCode();
        }
    }

这个类主要的作用是:

*  开启缓存刷新定时器;
 *  开启发送心跳定时器;
 *  开启实例instance状态变更监听;
 *  开启应用状态复制器.

### InstanceInfoReplicator ###

这个类是一个**Runnable**实际作用是用来更新和复制本地实例信息到远程服务器:

*  配置一个更新线程保证对远程服务器的更新;
 *  可以使用onDemandUpdate()按需调度任务;
 *  对任务的处理有速率限制;
 *  更新线程会被自动调用, 在第一次调用之后.

class InstanceInfoReplicator implements Runnable { 
    
        // first start 
        public void start(int initialDelayMs) { 
            if (started.compareAndSet(false, true)) { 
                instanceInfo.setIsDirty();  // for initial register
                Future next = scheduler.schedule(this, initialDelayMs, TimeUnit.SECONDS);
                scheduledPeriodicRef.set(next);
            }
        }
    
        // limit pull rate and fix mulit task
        // this function will be invoked by ApplicationInfoManager.StatusChangeListener()
        public boolean onDemandUpdate() { 
            ...
        }
        
        public void run() { 
            // refresh instance info
            discoveryClient.refreshInstanceInfo();
            Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
            if (dirtyTimestamp != null) { 
                // do register
                discoveryClient.register();
                instanceInfo.unsetIsDirty(dirtyTimestamp);
            }
            Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
            scheduledPeriodicRef.set(next);
        }
        
    }

### Client ###

这个类是用来帮助连接EurekaServer的.  
EurekaClient可以有一下几种使用方式:

*  向EurekaServer注册实例;
 *  向EurekaServer续约一个租期;
 *  向EurekaServer取消一个租期当服务从EurekaServer离开的时候;
 *  向EurekaServer查询注册在上面的服务/实例列表.

EurekaClient需要一个配置好的服务列表用来进行url请求. 默认情况下使用的是amazon的弹性api. 每一个api函数都应该有大量的容错处理进行考虑.

#### 整体脑图 ####

![f3109aa342ee08fe67ec0e6d88dca9d9.png][]

## Eureka-Server ##

在解读Eureka-Server端的源码时我们同样需要找到对应代码的入口.

根据start下的META-INF可以找到在启动时加载的类

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

在SpringCloud中他的入口并不是Netflix-Eureka中的init函数, 实际上Spring自己继承了**PeerAwareInstanceRegistryImpl**这个函数并且重写了其中的一些方法, 所以我们在开始阅读Eureka的源码之前先看一下**EurekaServerAutoConfiguration**这个类里面到底初始化了那些东西, 进入到这个类中可以看到SpringStart常见的启动套路这里我就不做过多的赘述了, 具体需要使用的Bean我放在了下面的脑图里.

### 自动注册 ###

### EurekaServerAutoConfiguration源码解读 ###

@Configuration
    @Import(EurekaServerInitializerConfiguration.class)
    @ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class)
    @EnableConfigurationProperties({  EurekaDashboardProperties.class,
    		InstanceRegistryProperties.class })
    @PropertySource("classpath:/eureka/server.properties")
    public class EurekaServerAutoConfiguration extends WebMvcConfigurerAdapter { 
    
        // Config eureka controller
    	@Bean
    	@ConditionalOnProperty(prefix = "eureka.dashboard", name = "enabled", matchIfMissing = true)
    	public EurekaController eurekaController() { ...}
        
        // Init peer eureka node for replace instances 
    	@Bean
    	@ConditionalOnMissingBean
    	public PeerEurekaNodes peerEurekaNodes(PeerAwareInstanceRegistry registry,
    			ServerCodecs serverCodecs,
                ReplicationClientAdditionalFilters replicationClientAdditionalFilters) { ...}
    
        // Eureka context ,It contains the @PostConstruct annotation , which annotates the logic 
        // will call back when the construct for the class has been loaded.
    	public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs,
                PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) { ...}
    
        // Init eureka server bootstarp , this class is an adpter for eureka bootstarp
        // will be invorked by EurekaServerInitializerConfiguration . Implemented in 
        // EurekaDerverInitializerConfiguration SmartLifecycle to asynchronous calls after 
        // the server start.
    	@Bean
    	public EurekaServerBootstrap eurekaServerBootstrap(PeerAwareInstanceRegistry registry,
                EurekaServerContext serverContext) { ...}
                
    	// Register the Jersey filter.
    	@Bean
    	public FilterRegistrationBean jerseyFilterRegistration(
    			javax.ws.rs.core.Application eurekaJerseyApp) { ...}

上面的代码是**EurekaServerAutoConfiguration**中比较重要的几个类, 内部的具体实现会在下面说明.  
这里先声明一下类上的注解:

1.  @Import(EurekaServerInitializerConfiguration.class): 这个注解非常重要,其中标示的类是主要加载EurekaServer的逻辑;
2.  @ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class): 标识一个有条件的bean只有匹配到这个类才回加载;
3.  @EnableConfigurationProperties(\{ EurekaDashboardProperties.class,
    
        InstanceRegistryProperties.class }): 声明配置类;
4.  @PropertySource(“classpath:/eureka/server.properties”): 注明配置文件的位置.

### EurekaServerInitializerConfiguration源码解读 ###

@Configuration(proxyBeanMethods = false)
    public class EurekaServerInitializerConfiguration
    		implements ServletContextAware, SmartLifecycle, Ordered { 
    
    	@Override
    	public void start() { 
            // start thread for asynchronous
    		new Thread(() -> { 
    			try { 
    				// init eureka server
    				eurekaServerBootstrap.contextInitialized(
    						EurekaServerInitializerConfiguration.this.servletContext);
                    log.info("Started Eureka Server");
                    // publish eureka server registry event
                    publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));
                    // set eureka is running
                    EurekaServerInitializerConfiguration.this.running = true;
                    // publish eureka server start event
    				publish(new EurekaServerStartedEvent(getEurekaServerConfig()));
    			}
    			catch (Exception ex) { 
    				// Help!
    				log.error("Could not initialize Eureka servlet context", ex);
    			}
    		}).start();
    	}

上文的注释中已经提到这个类的加载方式是通过SmartLifecycle进行加载的, 整理的主要逻辑是**eurekaServerBootstrap.contextInitialized(EurekaServerInitializerConfiguration.this.servletContext)** 这一行. 内部逻辑在下文中给出.

### EurekaServerBootstrap源码解读 ###

public class EurekaServerBootstrap { 
    
        public void contextInitialized(ServletContext context) { 
            
            initEurekaEnvironment();
            initEurekaServerContext();
    
            context.setAttribute(EurekaServerContext.class.getName(), this.serverContext);
        }
    
    	protected void initEurekaServerContext() throws Exception { 
    
    		EurekaServerContextHolder.initialize(this.serverContext);
    
    		log.info("Initialized server context");
    
    		// Copy registry from neighboring eureka node
            int registryCount = this.registry.syncUp();
            // in this function include:
            // Renewals happen every 30 seconds and for a minute it should be a factor of 2.
    		this.registry.openForTraffic(this.applicationInfoManager, registryCount);
    
    		// Register all monitoring statistics.
    		EurekaMonitors.registerAllStats();
    	}
    }

EurekaServerBootstrap主要加载环境比那辆以及context, 在最后\*\*initEuurekaServerContext()\*\*方法中区向其他eureka node 同步注册信息. 最后启动一个线程没三十秒同步两次.

### DefaultEurekaServerContext ###

@Singleton
    public class DefaultEurekaServerContext implements EurekaServerContext { 
        @PostConstruct
        @Override
        public void initialize() { 
    		// start one thread to read peer eureka node instances and per 60*10 second.
            logger.info("Initializing ...");
            peerEurekaNodes.start();
            try { 
                registry.init(peerEurekaNodes);
            } catch (Exception e) { 
                throw new RuntimeException(e);
            }
            logger.info("Initialized");
        }
    }

这个类是当所有的环境变量以及初始化的配置都准备好了之后加载初始化Eureka用的, 我们会对这个类进行详细的法分析.

> 与Eureka-Server的初始化方式不同, **EurekaBootStrap.contextInitialized()这个方法是监听到系统启动事件然后进行Eureka-Server的启动, 但是在Spring中所有的初始化动作是交给Bean处理的也就是说, 在启动时不会调用contextInitialized()这个方法, 而是直接经过Bean的注入后DefaultEurekaServerContext.initialize()方法实际是因为@PostConstruct**注解在构造方法调用结束后执行的逻辑

### PeerAwareInstanceRegistryImpl ###

@Singleton
    public class PeerAwareInstanceRegistryImpl extends AbstractInstanceRegistry implements PeerAwareInstanceRegistry { 
        @Override
        public void init(PeerEurekaNodes peerEurekaNodes) throws Exception { 
            this.numberOfReplicationsLastMin.start();
            this.peerEurekaNodes = peerEurekaNodes;
            initializedResponseCache();
    		// update renewal threshold per 15 minutes
            scheduleRenewalThresholdUpdateTask();
    		// 
            initRemoteRegionRegistry();
        }
    }

### 资源管理 ###

作为一个服务端是需要对外提供接口的, 实际上Eureka-Server本质上就是一个服务, 作为服务端他提供了三种类型的访问:

*  ApplicationResource: 用于app应用级别的资源控制;
 *  InstanceResource: 用于instance实例级别的资源控制;
 *  PeerReplicationResource: 用于peer eureka server之间的数据同步;

#### ApplicationResource ####

@Produces({ "application/xml", "application/json"})
    public class ApplicationResource { 
        @POST
        @Consumes({ "application/json", "application/xml"})
        public Response addInstance(InstanceInfo info,
                                    @HeaderParam(PeerEurekaNode.HEADER_REPLICATION) String isReplication) { 
                                        ...
            registry.register(info, "true".equals(isReplication));                      }
    }

#### 整体脑图 ####

![5452541c29d50446a4e87b525f3cf837.png][]

## 线上问题复现 ##

### 出现场景 ###

我们的线上环境和预发环境均可以复现这个问题, 这里描述一下集群场景, 使用的集群为3台主从模式的构建方式和.  
集群上所注册的服务总数超过1000个(这个1000的数值到底哪里特殊目前没有弄清楚).  
有一个服务使用相同的instanceId和appName进行注册数量超过10个.  
在这些条件都满足的情况下会出现eureka-server将服务丢失的情况.

### 场景复现 ###

在本地环境我同样使用3台的主从集群, 然后模拟1300个不同的服务进行registry和renew然后在模拟1300个请求使用同样的instanceId进行registry和renew.

这种情况下eureka-server运行良好, 并没有出现任何问题.

[Link 1]: https://github.com/Netflix/eureka/wiki/Eureka-at-a-glance
[e87af352e5a889951fa8d368c7d46fd4.png]: /images/20221120/445f1d9fd92241f4ba446150915d9f22.png
[3ee0caeec9aaa716f3dcfdc805dbd38e.png]: https://img-blog.csdnimg.cn/img_convert/3ee0caeec9aaa716f3dcfdc805dbd38e.png
[f3109aa342ee08fe67ec0e6d88dca9d9.png]: https://img-blog.csdnimg.cn/img_convert/f3109aa342ee08fe67ec0e6d88dca9d9.png
[5452541c29d50446a4e87b525f3cf837.png]: https://img-blog.csdnimg.cn/img_convert/5452541c29d50446a4e87b525f3cf837.png