SpringCloud 源码系列（1）—— 注册中心 Eureka（上）

<?xml version="1.0" encoding="UTF-8"?>
web-app version="2.5"
 3          xmlns="http://java.sun.com/xml/ns/javaee"
 4          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 5          xsi:schemaLocation="http://java.sun.com/xml/ns/javaee
    http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd" 7   <!-- eureka 启动初始化类 -->
 8   listener 9     listener-class>com.netflix.eureka.EurekaBootStrap10   11 
12    状态过滤器 13   filter14     filter-name>statusFilter15     filter-class>com.netflix.eureka.StatusFilter16   17 
18    认证过滤器 19   20     >requestAuthFilter21     >com.netflix.eureka.ServerRequestAuthFilter22   23 
24    限流过滤器 25   26     >rateLimitingFilter27     >com.netflix.eureka.RateLimitingFilter28   29   30     >gzipEncodingEnforcingFilter31     >com.netflix.eureka.GzipEncodingEnforcingFilter32   33 
34    jersey 容器 35   36     >jersey37     >com.sun.jersey.spi.container.servlet.ServletContainer38     init-param39       param-name>com.sun.jersey.config.property.WebPageContentRegex40       param-value>/(flex|images|js|css|jsp)/.*41     42     43       >com.sun.jersey.config.property.packages44       >com.sun.jersey;com.netflix45     46 
47      GZIP content encoding/decoding 48     49       >com.sun.jersey.spi.container.ContainerRequestFilters50       >com.sun.jersey.api.container.filter.GZIPContentEncodingFilter51     52     53       >com.sun.jersey.spi.container.ContainerResponseFilters54       55     56   57 
58   filter-mapping59     60     url-pattern>/*61   62 
63   64     65     66   67 
68    Uncomment this to enable rate limiter filter.
69   <filter-mapping>
70     <filter-name>rateLimitingFilter</filter-name>
71     <url-pattern>/v2/apps</url-pattern>
72     <url-pattern>/v2/apps/*</url-pattern>
73   </filter-mapping>
74   75 
76   77     78     >/v2/apps79     >/v2/apps/*80   81 
82   83     84     85   86 
87    欢迎页 88   welcome-file-list89     welcome-file>jsp/status.jsp90   91 
92 web-app>

synchronized InstanceInfo get() {
if (instanceInfo ==  3          续约信息：主要有续约间隔时间（默认30秒）和续约过期时间（默认90秒）
 4         LeaseInfo.Builder leaseInfoBuilder = LeaseInfo.Builder.newBuilder()
                .setRenewalIntervalInSecs(config.getLeaseRenewalIntervalInSeconds())
                .setDurationInSecs(config.getLeaseExpirationDurationInSeconds());
 7 
if (vipAddressResolver ==  9             vipAddressResolver =  Archaius1VipAddressResolver();
        }
12          基于建造者模式来创建 InstanceInfo
13         InstanceInfo.Builder builder = InstanceInfo.Builder.newBuilder(vipAddressResolver);
15          set the appropriate id for the InstanceInfo,falling back to datacenter Id if applicable,else hostname
16         String instanceId = config.getInstanceId();
17         if (instanceId == null || instanceId.isEmpty()) {
18             DataCenterInfo dataCenterInfo = config.getDataCenterInfo();
19             if (dataCenterInfo instanceof UniqueIdentifier) {
20                 instanceId = ((UniqueIdentifier) dataCenterInfo).getId();
21             } 22                 instanceId = config.getHostName(23             }
24         String defaultAddress;
27         if (config  RefreshableInstanceConfig) {
28              Refresh AWS data center info,and return up to date address
29             defaultAddress = ((RefreshableInstanceConfig) config).resolveDefaultAddress(30         } 31             defaultAddress = config.getHostName(34          fail safe
35         if (defaultAddress ==  defaultAddress.isEmpty()) {
36             defaultAddress = config.getIpAddress();
38 
39          设置属性
        builder.setNamespace(config.getNamespace())
                .setInstanceId(instanceId)
42                 .setAppName(config.getAppname())
                .setAppGroupName(config.getAppGroupName())
44                 .setDataCenterInfo(config.getDataCenterInfo())
45                 .setIPAddr(config.getIpAddress())
46                 .setHostName(defaultAddress)
                .setPort(config.getNonSecurePort())
                .enablePort(PortType.UNSECURE,config.isNonSecurePortEnabled())
                .setSecurePort(config.getSecurePort())
                .enablePort(PortType.SECURE,config.getSecurePortEnabled())
                .setVIPAddress(config.getVirtualHostName())
                .setSecureVIPAddress(config.getSecureVirtualHostName())
53                 .setHomePageUrl(config.getHomePageUrlPath(),config.getHomePageUrl())
54                 .setStatusPageUrl(config.getStatusPageUrlPath(),config.getStatusPageUrl())
55                 .setASGName(config.getASGName())
                .setHealthCheckUrls(config.getHealthCheckUrlPath(),1)">                        config.getHealthCheckUrl(),config.getSecureHealthCheckUrl());
58 
59 
60          Start off with the STARTING state to avoid traffic
61         if (!config.isInstanceEnabledOnit()) {
62             InstanceStatus initialStatus = InstanceStatus.STARTING;
63             LOG.info("Setting initial instance status as: {}"64             builder.setStatus(initialStatus);
65         } 66             LOG.info("Setting initial instance status as: {}. This may be too early for the instance to advertise "
67                      + "itself as available. You would instead want to control this via a healthcheck handler."                     InstanceStatus.UP);
70 
71          Add any user-specific metadata information
72         for (Map.Entry<String,String> mapEntry : config.getMetadataMap().entrySet()) {
73             String key = mapEntry.getKey();
74             String value = mapEntry.getValue();
75              only add the metadata if the value is present
76             if (value != null && !value.isEmpty()) {
                builder.add(key,value);
78 79 80 
81          调用 build 方法做属性校验并创建 InstanceInfo 实例
82         instanceInfo = builder.build();
        instanceInfo.setLeaseInfo(leaseInfoBuilder.build());
84  instanceInfo;
86 }

  1 DiscoveryClient(ApplicationInfoManager applicationInfoManager,EurekaClientConfig config,AbstractDiscoveryClientOptionalArgs args,1)">  2                 Provider<BackupRegistry> backupRegistryProvider,EndpointRandomizer endpointRandomizer) {
  3     if (args !=   4         this.healthCheckHandlerProvider = args.healthCheckHandlerProvider;
  5         this.healthCheckCallbackProvider = args.healthCheckCallbackProvider;
  6         this.eventListeners.addAll(args.getEventListeners());
  7         this.preRegistrationHandler = args.preRegistrationHandler;
  8     }   9         this.healthCheckCallbackProvider =  10         this.healthCheckHandlerProvider =  11         this.preRegistrationHandler =  12  13 
 14      将实例信息、配置信息保存到本地
 15     this.applicationInfoManager = applicationInfoManager;
 16     InstanceInfo myInfo = applicationInfoManager.getInfo();
 17     clientConfig = config;
 18     staticClientConfig = clientConfig;
 19     transportConfig = config.getTransportConfig();
 20     instanceInfo = myInfo;
 21     if (myInfo !=  22         appPathIdentifier = instanceInfo.getAppName() + "/" + instanceInfo.getId();
 23     }  24         logger.warn("Setting instanceInfo to a passed in null value" 25  26 
 27     this.backupRegistryProvider = backupRegistryProvider;
 28     this.endpointRandomizer = endpointRandomizer;
 29     this.urlRandomizer =  EndpointUtils.InstanceInfoBasedUrlRandomizer(instanceInfo);
 30     localRegionApps.set( Applications());
 31 
 32 
 33     fetchRegistryGeneration = new AtomicLong(0 34     remoteRegionsToFetch = new AtomicReference<String>(clientConfig.fetchRegistryForRemoteRegions());
 35      从远程拉取注册表的地址数组，使用的原子类，在运行中可能会动态更新地址
 36     remoteRegionsRef = new AtomicReference<>(remoteRegionsToFetch.get() == null ? null : remoteRegionsToFetch.get().split(","));
 37 
 38      如果要获取注册表，就会注册状态监视器
 39      (config.shouldFetchRegistry()) {
 40         this.registryStalenessMonitor = new ThresholdLevelsMetric(this,METRIC_REGISTRY_PREFIX + "lastUpdateSec_",1)">new long[]{15L,30L,60L,120L,240L,480L});
 41     }  42         this.registryStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
 43  44 
 45      如果要注册到 eureka-server，就会创建心跳状态监视器
 46      (config.shouldRegisterWithEureka()) {
 47         this.heartbeatStalenessMonitor =  48     }  49         this.heartbeatStalenessMonitor = 50  51 
 52     logger.info("Initializing Eureka in region {}" 53 
 54      如果不注册到注册中心，且不拉取注册表，就不创建调度器、线程池等资源了
 55     if (!config.shouldRegisterWithEureka() && !config.shouldFetchRegistry()) {
 56         logger.info("Client configured to neither register nor query for data." 57         scheduler =  58         heartbeatExecutor =  59         cacheRefreshExecutor =  60         eurekaTransport =  61         instanceRegionChecker = new InstanceRegionChecker( PropertyBasedAzToRegionMapper(config),1)"> 62 
 63          This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
 64          to work with DI'd DiscoveryClient
 65         DiscoveryManager.getInstance().setDiscoveryClient( 66         DiscoveryManager.getInstance().setEurekaClientConfig(config);
 67 
 68         initTimestampMs = System.currentTimeMillis();
 69         initRegistrySize = .getApplications().size();
 70         registrySize = initRegistrySize;
 71         logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}" 72                 initTimestampMs,initRegistrySize);
 73 
 74         return;   no need to setup up an network tasks and we are done
 75  76 
 77      78          创建定时调度器，默认有2个核心线程，主要处理心跳任务和缓存刷新任务
 79         scheduler = Executors.newScheduledThreadPool(2 80                  ThreadFactoryBuilder()
 81                         .setNameFormat("DiscoveryClient-%d" 82                         .setDaemon(true 83                         .build());
 84 
 85          维持心跳的线程池，一个核心线程，最大线程数默认5。
 86          注意其使用的队列是 SynchronousQueue 队列，这个队列只能放一个任务，一个线程将任务取走后，才能放入下一个任务，否则只能阻塞。
 87         heartbeatExecutor =  ThreadPoolExecutor(
 88                 1,clientConfig.getHeartbeatExecutorThreadPoolSize(),0 89                 new SynchronousQueue<Runnable>(),1)"> 90                  91                         .setNameFormat("DiscoveryClient-HeartbeatExecutor-%d" 92                         .setDaemon( 93                         .build()
 94         );   use direct handoff
 95 
 96          刷新缓存的线程池，一个核心线程，最大线程数据默认为5
 97         cacheRefreshExecutor =  98                 1,clientConfig.getCacheRefreshExecutorThreadPoolSize(),1)"> 99                 100                 101                         .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d"102                         .setDaemon(103 104         );  105 
106          eureka http 调用客户端，支持 eureka client 与 eureka server 之间的通信
107         eurekaTransport =  EurekaTransport();
108          初始化 eurekaTransport
109         scheduleServerEndpointTask(eurekaTransport,1)">110 
111         AzToRegionMapper azToRegionMapper;
112          (clientConfig.shouldUseDnsForFetchingServiceUrls()) {
113             azToRegionMapper =  DNSBasedAzToRegionMapper(clientConfig);
114         } 115             azToRegionMapper =  PropertyBasedAzToRegionMapper(clientConfig);
116 117         if (null != remoteRegionsToFetch.get()) {
118             azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(",1)">119 120         instanceRegionChecker =  InstanceRegionChecker(azToRegionMapper,1)">121     } 122         new RuntimeException("Failed to initialize DiscoveryClient!"123 124 
125      (clientConfig.shouldFetchRegistry()) {
126         127              拉取注册表：全量抓取和增量抓取
128             boolean primaryFetchRegistryResult = fetchRegistry(129             primaryFetchRegistryResult) {
130                 logger.info("Initial registry fetch from primary servers failed"131 132             boolean backupFetchRegistryResult = 133             if (!primaryFetchRegistryResult && !fetchRegistryFromBackup()) {
134                 backupFetchRegistryResult = 135                 logger.info("Initial registry fetch from backup servers failed"136 137             if (!primaryFetchRegistryResult && !backupFetchRegistryResult && clientConfig.shouldEnforceFetchRegistryAtInit()) {
138                 new IllegalStateException("Fetch registry error at startup. Initial fetch failed."139 140         }  (Throwable th) {
141             logger.error("Fetch registry error at startup: {}"142              IllegalStateException(th);
143 144 145 
146      call and execute the pre registration handler before all background tasks (inc registration) is started
147     this.preRegistrationHandler != 148         .preRegistrationHandler.beforeRegistration();
149 150 
151     if (clientConfig.shouldRegisterWithEureka() && clientConfig.shouldEnforceRegistrationAtInit()) {
152         153             register() ) {
154                 new IllegalStateException("Registration error at startup. Invalid server response."155 156         } 157             logger.error("Registration error at startup: {}"158             159 160 161 
162      初始化一些调度任务：刷新缓存的调度任务、发送心跳的调度任务、实例副本传播器
163     initScheduledTasks();
164 
165     166         Monitors.registerObject(167     } 168         logger.warn("Cannot register timers"169 170 
171     172     173     DiscoveryManager.getInstance().setDiscoveryClient(174     DiscoveryManager.getInstance().setEurekaClientConfig(config);
175 
176      初始化的时间
177     initTimestampMs =178     initRegistrySize = 179     registrySize =180     logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}"181             initTimestampMs,1)">182 183 
184 ////////////////////////////////////////////////////////////////////
185 
186  initScheduledTasks() {
187     188          抓取注册表的间隔时间，默认30秒
189         int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
190          刷新缓存调度器延迟时间扩大倍数，在任务超时的时候，将扩大延迟时间
191          这在出现网络抖动、eureka-sever 不可用时，可以避免频繁发起无效的调度
192         int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
193          注册表刷新的定时任务
194         cacheRefreshTask =  TimedSupervisorTask(
195                 "cacheRefresh"196                 scheduler,1)">197                 cacheRefreshExecutor,1)">198                 registryFetchIntervalSeconds,1)">199                 TimeUnit.SECONDS,1)">200                 expBackOffBound,1)">201                 new CacheRefreshThread()  刷新注册表的任务
202 203          30秒后开始调度刷新注册表的任务
204         scheduler.schedule(
205                 cacheRefreshTask,1)">206 207 208 
209      (clientConfig.shouldRegisterWithEureka()) {
210          续约间隔时间，默认30秒
211         int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
212          心跳调度器的延迟时间扩大倍数，默认10
213          clientConfig.getHeartbeatExecutorExponentialBackOffBound();
214         logger.info("Starting heartbeat executor: " + "renew interval is: {}"215 
216          心跳的定时任务
217         heartbeatTask = 218                 "heartbeat"219 220                 heartbeatExecutor,1)">221                 renewalIntervalInSecs,1)">222 223 224                  HeartbeatThread()
225 226          30秒后开始调度心跳的任务
227 228                 heartbeatTask,1)">229 230 
231          实例副本传播器，用于定时更新自己状态
232         instanceInfoReplicator =  InstanceInfoReplicator(
233                 234                 instanceInfo,1)">235                 clientConfig.getInstanceInfoReplicationIntervalSeconds(),1)">236                 2);  burstSize
237 
238          实例状态变更的监听器
239         statusChangeListener =  ApplicationInfoManager.StatusChangeListener() {
240             @Override
241              String getId() {
242                 return "statusChangeListener"243 244 
245 246              notify(StatusChangeEvent statusChangeEvent) {
247                 if (statusChangeEvent.getStatus() == InstanceStatus.DOWN) {
248                     logger.error("Saw local status change event {}"249                 } 250                     logger.info("Saw local status change event {}"251                 }
252                 instanceInfoReplicator.onDemandUpdate();
253 254         };
255 
256          向 ApplicationInfoManager 注册监听器
257          (clientConfig.shouldOnDemandUpdateStatusChange()) {
258             applicationInfoManager.registerStatusChangeListener(statusChangeListener);
259 260 
261          启动副本传播器，默认延迟时间40秒
262         instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
263     } 264         logger.info("Not registering with Eureka server per configuration"265 266 }

        EurekaServerConfig serverConfig,1)">        EurekaClientConfig clientConfig,1)">        ServerCodecs serverCodecs,1)">        EurekaClient eurekaClient
super(serverConfig,clientConfig,serverCodecs);
this.eurekaClient = eurekaClient;
 最近一分钟集群同步的次数计数器
this.numberOfReplicationsLastMin = new MeasuredRate(1000 * 60 * 1 We first check if the instance is STARTING or DOWN,then we check explicit overrides,1)"> then we check the status of a potentially existing lease.
13     this.instanceStatusOverrideRule = new FirstMatchWinsCompositeRule( DownOrStartingRule(),1)">14             new OverrideExistsRule(overriddenInstanceStatusMap),1)"> LeaseExistsRule());
16 
///////////////////////////////////////////////
19 
21 protected AbstractInstanceRegistry(EurekaServerConfig serverConfig,EurekaClientConfig clientConfig,ServerCodecs serverCodecs) {
22     this.serverConfig = serverConfig;
23     this.clientConfig =24     this.serverCodecs = serverCodecs;
25      最近下线的循环队列
this.recentCanceledQueue = new CircularQueue<Pair<Long,String>>(1000 最近注册的循环队列
28     this.recentRegisteredQueue = 29 
 最近一分钟续约的计数器
this.renewsLastMin = 32      一个定时调度任务，定时剔除最近改变队列中过期的实例
33     .deltaRetentionTimer.schedule(getDeltaRetentionTask(),1)">            serverConfig.getDeltaRetentionTimerIntervalInMs(),1)">35             serverConfig.getDeltaRetentionTimerIntervalInMs());
36 }

 start() {
 单个线程的线程池
 3     taskExecutor = Executors.newSingleThreadScheduledExecutor(
 4              ThreadFactory() {
                @Override
 6                  Thread newThread(Runnable r) {
 7                     Thread thread = new Thread(r,"Eureka-PeerNodesUpdater" 8                     thread.setDaemon( 9                      thread;
 根据集群地址更新 PeerEurekaNode，PeerEurekaNode 就包含了调度其它注册中心的客户端
        updatePeerEurekaNodes(resolvePeerUrls());
16         Runnable peersUpdateTask =  Runnable() {
18             19                                     updatePeerEurekaNodes(resolvePeerUrls());
21                 } 22                     logger.error("Cannot update the replica Nodes" 定时跟新集群信息 PeerEurekaNode，如果有eureka-server不可用了，就可以及时下线，或者新上线了eureka-server，可以及时感知到
        taskExecutor.scheduleWithFixedDelay(
                peersUpdateTask,1)">                serverConfig.getPeerEurekaNodesUpdateIntervalMs(),1)">31                 TimeUnit.MILLISECONDS
33 34     }  IllegalStateException(e);
36 for (PeerEurekaNode node : peerEurekaNodes) {
38         logger.info("Replica node URL:  {}"40 }

void init(PeerEurekaNodes peerEurekaNodes)  启动计数器
.numberOfReplicationsLastMin.start();
this.peerEurekaNodes = peerEurekaNodes;
 初始化响应缓存，eureka server 构造了一个多级缓存来响应客户端抓取注册表的请求
    initializedResponseCache();
 定时调度任务更新续约阀值，主要就是更新 numberOfRenewsPerMinThreshold 这个值，即每分钟续约次数
    scheduleRenewalThresholdUpdateTask();
 初始化 RemoteRegionRegistry
    initRemoteRegionRegistry();
13         Monitors.registerObject(15         logger.warn("Cannot register the JMX monitor for the InstanceRegistry :"17 }

 syncUp() {
 Copy entire entry from neighboring DS node
int count = 0 注册表同步重试次数，默认5次
for (int i = 0; ((i < serverConfig.getRegistrySyncRetries()) && (count == 0)); i++if (i > 0 8              9                  同步重试时间，默认30秒
                Thread.sleep(serverConfig.getRegistrySyncRetryWaitMs());
11             }  (InterruptedException e) {
12                 logger.warn("Interrupted during registry transfer.."13                 break16         Applications apps = eurekaClient.getApplications();
 (Application app : apps.getRegisteredApplications()) {
 (InstanceInfo instance : app.getInstances()) {
20                      (isRegisterable(instance)) {
21                          注册实例
22                         register(instance,instance.getLeaseInfo().getDurationInSecs(),1)">23                         count++                    }
25                 }  (Throwable t) {
26                     logger.error("During DS init copy" count;
32 }

void openForTraffic(ApplicationInfoManager applicationInfoManager,1)"> count) {
 期望的客户端每分钟的续约次数
this.expectedNumberOfClientsSendingRenews = 更新每分钟续约阀值
    updateRenewsPerMinThreshold();
 7     logger.info("Got {} instances from neighboring DS node" 8     logger.info("Renew threshold is: {}"this.startupTime =if (count > 011         this.peerInstancesTransferEmptyOnStartup = 13     DataCenterInfo.Name selfName = applicationInfoManager.getInfo().getDataCenterInfo().getName();
boolean isAws = Name.Amazon == selfName;
if (isAws && serverConfig.shouldPrimeAwsReplicaConnections()) {
16         logger.info("Priming AWS connections for all replicas.."        primeAwsReplicas(applicationInfoManager);
19     logger.info("Changing status to UP" 设置实例状态为已启动
    applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
.postInit();
//////////////////////////////////////26 
 postInit() {
 启动 统计最近一分钟续约次数的计数器
    renewsLastMin.start();
if (evictionTaskRef.get() !=         evictionTaskRef.get().cancel();
 定时剔除任务
34     evictionTaskRef.set( EvictionTask());
    evictionTimer.schedule(evictionTaskRef.get(),1)">            serverConfig.getEvictionIntervalTimerInMs(),1)">            serverConfig.getEvictionIntervalTimerInMs());
38 }