Fork me on GitHub

framework/SpringCloud/eureka/Eureka

字数统计: 14.3k字   |   阅读时长≈ 73分

Eureka

客户端

@EnableDiscoveryClient

它只做了一些事，就是把一个配置类加载进来，这个配置类 有一个 主动注册 ，注册管理的属性配置，先记着，以后用的到

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

/*
@Import(EnableDiscoveryClientImportSelector.class)

这个注解是专门用于导入第三方bean信息的，但是如果他是一个 ImportSelector类型，则会导入  selectImports(AnnotationMetadata metadata)返回的string数组 ，这个string数组就是一堆类的全路径

metadata这个参数 就是 这个注册所在的配置类的信息


*/

@EnableDiscoveryClient这个注解的内容如下：

@Import(EnableDiscoveryClientImportSelector.class)
public @interface EnableDiscoveryClient {

	/**
	 * 如果为true，ServiceRegistry将自动注册本地服务器。
	 * @return - {@code true} if you want to automatically register.
	 */
	boolean autoRegister() default true;

}

EnableDiscoveryClientImportSelector主要内容如下
    @Override
	public String[] selectImports(AnnotationMetadata metadata) {
		String[] imports = super.selectImports(metadata);//获取spring.factory配置的key为这个注解全路径，的value取出来

    //获取这个注解属性 变成一个map返回
		AnnotationAttributes attributes = AnnotationAttributes.fromMap(
				metadata.getAnnotationAttributes(getAnnotationClass().getName(), true));

    //因为上面默认是true，所以autoRegister为true
		boolean autoRegister = attributes.getBoolean("autoRegister");

		if (autoRegister) {
			List<String> importsList = new ArrayList<>(Arrays.asList(imports));
            //加上一个AutoServiceRegistrationConfiguration配置类，以后spring容器会初始化这个bean
			importsList.add(
					"org.springframework.cloud.client.serviceregistry.AutoServiceRegistrationConfiguration");
			imports = importsList.toArray(new String[0]);
		}
		else {
			Environment env = getEnvironment();
			if (ConfigurableEnvironment.class.isInstance(env)) {
				ConfigurableEnvironment configEnv = (ConfigurableEnvironment) env;
				LinkedHashMap<String, Object> map = new LinkedHashMap<>();
				map.put("spring.cloud.service-registry.auto-registration.enabled", false);
				MapPropertySource propertySource = new MapPropertySource(
						"springCloudDiscoveryClient", map);
				configEnv.getPropertySources().addLast(propertySource);
			}

		}

		return imports;
	}

/*
@ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled",
		matchIfMissing = true)
这个注解最后返回true就代表满足条件


*/
@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties(AutoServiceRegistrationProperties.class)
@ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled",
		matchIfMissing = true)
public class AutoServiceRegistrationConfiguration {

}



//就三个属性值配置
@ConfigurationProperties("spring.cloud.service-registry.auto-registration")
public class AutoServiceRegistrationProperties {

	/** Whether service auto-registration is enabled. Defaults to true. */
	private boolean enabled = true;

	/** Whether to register the management as a service. Defaults to true. */
	private boolean registerManagement = true;

	/**
	 * Whether startup fails if there is no AutoServiceRegistration. Defaults to false.
	 */
	private boolean failFast = false;

}

查看spring-cloud-netflix-eureka-client包下 meta-inf/ spring.factory文件

1
2
3
4
5
6
7
8
9
10
11
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.netflix.eureka.config.EurekaClientConfigServerAutoConfiguration,\
org.springframework.cloud.netflix.eureka.config.DiscoveryClientOptionalArgsConfiguration,\
org.springframework.cloud.netflix.eureka.EurekaClientAutoConfiguration,\
org.springframework.cloud.netflix.ribbon.eureka.RibbonEurekaAutoConfiguration,\
org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration,\
org.springframework.cloud.netflix.eureka.reactive.EurekaReactiveDiscoveryClientConfiguration,\
org.springframework.cloud.netflix.eureka.loadbalancer.LoadBalancerEurekaAutoConfiguration

org.springframework.cloud.bootstrap.BootstrapConfiguration=\
org.springframework.cloud.netflix.eureka.config.EurekaConfigServerBootstrapConfiguration

DiscoveryClientOptionalArgsConfiguration

注册一个客户端的 可选参数，主要用于 去setTransportClientFactories(new RestTemplateTransportClientFactories());

就是设置一个 用于 创建客户端通信对象的工厂

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
//第一个bean是 用于使用RestTemplate去传输集群间通信请求的
//但是有一个条件，就是不能存在com.sun.jersey.api.client.filter.ClientFilter这个过滤器的类
    @Bean
	@ConditionalOnMissingClass("com.sun.jersey.api.client.filter.ClientFilter")
	@ConditionalOnMissingBean(value = { AbstractDiscoveryClientOptionalArgs.class },
			search = SearchStrategy.CURRENT)
	@ConditionalOnProperty(prefix = "eureka.client", name = "webclient.enabled",
			matchIfMissing = true, havingValue = "false")
	public RestTemplateDiscoveryClientOptionalArgs restTemplateDiscoveryClientOptionalArgs() {
		logger.info("Eureka HTTP Client uses RestTemplate.");
		return new RestTemplateDiscoveryClientOptionalArgs();
	}


//第二个bean是使用 react的webclient进行集群间通信，前提也是com.sun.jersey.api.client.filter.ClientFilter它不存再得情况下
	@Bean
	@ConditionalOnMissingClass("com.sun.jersey.api.client.filter.ClientFilter")
	@ConditionalOnClass(
			name = "org.springframework.web.reactive.function.client.WebClient")
	@ConditionalOnMissingBean(
			value = { AbstractDiscoveryClientOptionalArgs.class,
					RestTemplateDiscoveryClientOptionalArgs.class },
			search = SearchStrategy.CURRENT)
	@ConditionalOnProperty(prefix = "eureka.client", name = "webclient.enabled",
			havingValue = "true")
	public WebClientDiscoveryClientOptionalArgs webClientDiscoveryClientOptionalArgs() {
		logger.info("Eureka HTTP Client uses WebClient.");
		return new WebClientDiscoveryClientOptionalArgs();
	}


//这个是com.sun.jersey.api.client.filter.ClientFilter存在，则加载
	@Bean
	@ConditionalOnClass(name = "com.sun.jersey.api.client.filter.ClientFilter")
	@ConditionalOnMissingBean(value = AbstractDiscoveryClientOptionalArgs.class,
			search = SearchStrategy.CURRENT)
	public MutableDiscoveryClientOptionalArgs discoveryClientOptionalArgs() {
		return new MutableDiscoveryClientOptionalArgs();
	}


  //最后一个是异常，不用管
//大概意思就是 配置了eureka.client.webclient.enabled=true，但是"org.springframework.web.reactive.function.client.WebClient"没有存在于classpath中
	@Configuration
	@ConditionalOnMissingClass({ "com.sun.jersey.api.client.filter.ClientFilter",
			"org.springframework.web.reactive.function.client.WebClient" })
	@ConditionalOnProperty(prefix = "eureka.client", name = "webclient.enabled",
			havingValue = "true")
	protected static class WebClientNotFoundConfiguration {

		public WebClientNotFoundConfiguration() {
			throw new IllegalStateException("eureka.client.webclient.enabled is true, "
					+ "but WebClient is not on the classpath. Please add "
					+ "spring-boot-starter-webflux as a dependency.");
		}

	}

RefreshAutoConfiguration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
	//一个刷新作用域对象
    @Bean
	@ConditionalOnMissingBean(RefreshScope.class)
	public static RefreshScope refreshScope() {
		return new RefreshScope();
	}

    //一个监听环境变化的 日志记录器
    @Bean
	@ConditionalOnMissingBean
	public static LoggingRebinder loggingRebinder() {
		return new LoggingRebinder();
	}

//初始化一个用于 上下文刷新的 对象
	@Bean
	@ConditionalOnMissingBean
	public ContextRefresher contextRefresher(ConfigurableApplicationContext context,
			RefreshScope scope) {
		return new ContextRefresher(context, scope);
	}


//初始化一个 刷新事件的监听器
    @Bean
	public RefreshEventListener refreshEventListener(ContextRefresher contextRefresher) {
		return new RefreshEventListener(contextRefresher);
	}

//对刷新作用域bean信息的增强器
@Component
	protected static class RefreshScopeBeanDefinitionEnhancer
			implements BeanDefinitionRegistryPostProcessor, EnvironmentAware {
        
      public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry registry)
				throws BeansException {
			bindEnvironmentIfNeeded(registry);
			for (String name : registry.getBeanDefinitionNames()) {
				BeanDefinition definition = registry.getBeanDefinition(name);
				if (isApplicable(registry, name, definition)) {
					BeanDefinitionHolder holder = new BeanDefinitionHolder(definition,
							name);
					BeanDefinitionHolder proxy = ScopedProxyUtils
							.createScopedProxy(holder, registry, true);
					definition.setScope("refresh");
					if (registry.containsBeanDefinition(proxy.getBeanName())) {
						registry.removeBeanDefinition(proxy.getBeanName());
					}
					registry.registerBeanDefinition(proxy.getBeanName(),
							proxy.getBeanDefinition());
				}
			}
		}  
        
    }

EurekaDiscoveryClientConfiguration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties
@ConditionalOnClass(EurekaClientConfig.class)
@ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true)
@ConditionalOnDiscoveryEnabled
@ConditionalOnBlockingDiscoveryEnabled


//初始化一个 封装了EurekaClient和EurekaClientConfig的 EurekaDiscoveryClient类的对象
    @Bean
	@ConditionalOnMissingBean
	public EurekaDiscoveryClient discoveryClient(EurekaClient client,
			EurekaClientConfig clientConfig) {
		return new EurekaDiscoveryClient(client, clientConfig);
	}


   @Configuration(proxyBeanMethods = false)
	@ConditionalOnProperty(value = "eureka.client.healthcheck.enabled",
			matchIfMissing = false)
	protected static class EurekaHealthCheckHandlerConfiguration {

		@Autowired(required = false)
		private StatusAggregator statusAggregator = new SimpleStatusAggregator();

        //提供一个eureka状态检查处理
		@Bean
		@ConditionalOnMissingBean(HealthCheckHandler.class)
		public EurekaHealthCheckHandler eurekaHealthCheckHandler() {
			return new EurekaHealthCheckHandler(this.statusAggregator);
		}

	}



    @Configuration(proxyBeanMethods = false)
	@ConditionalOnClass(RefreshScopeRefreshedEvent.class)
	protected static class EurekaClientConfigurationRefresher
			implements ApplicationListener<RefreshScopeRefreshedEvent> {

		@Autowired(required = false)
		private EurekaClient eurekaClient;

		@Autowired(required = false)
		private EurekaAutoServiceRegistration autoRegistration;

		public void onApplicationEvent(RefreshScopeRefreshedEvent event) {
			// This will force the creation of the EurkaClient bean if not already created
			// to make sure the client will be reregistered after a refresh event
			if (eurekaClient != null) {
				eurekaClient.getApplications();
			}
			if (autoRegistration != null) {
				// register in case meta data changed
				this.autoRegistration.stop();
				this.autoRegistration.start();
			}
		}

	}

AutoServiceRegistrationAutoConfiguration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

/*
没干什么事情
*/
@Configuration(proxyBeanMethods = false)
@Import(AutoServiceRegistrationConfiguration.class)
@ConditionalOnProperty(value = "spring.cloud.service-registry.auto-registration.enabled",
		matchIfMissing = true)
public class AutoServiceRegistrationAutoConfiguration {

	@Autowired(required = false)
	private AutoServiceRegistration autoServiceRegistration;

   
	@Autowired
	private AutoServiceRegistrationProperties properties;

	@PostConstruct
	protected void init() {
		if (this.autoServiceRegistration == null && this.properties.isFailFast()) {
			throw new IllegalStateException("Auto Service Registration has "
					+ "been requested, but there is no AutoServiceRegistration bean");
		}
	}

}

EurekaClientAutoConfiguration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124


@Configuration(proxyBeanMethods = false)
@EnableConfigurationProperties//没指定什么
@ConditionalOnClass(EurekaClientConfig.class)//必须存在EurekaClientConfig类
@ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true)//默认是true

@ConditionalOnDiscoveryEnabled//默认是true

//本配置类加载完成后，才会加载指定的类
@AutoConfigureBefore({ NoopDiscoveryClientAutoConfiguration.class,
		CommonsClientAutoConfiguration.class, ServiceRegistryAutoConfiguration.class })

//本配置类加载前，会提前加载指定的类
@AutoConfigureAfter(name = {
		"org.springframework.cloud.netflix.eureka.config.DiscoveryClientOptionalArgsConfiguration",
		"org.springframework.cloud.autoconfigure.RefreshAutoConfiguration",
		"org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration",
		"org.springframework.cloud.client.serviceregistry.AutoServiceRegistrationAutoConfiguration" })



//不知干什么用的
    @Bean
	public HasFeatures eurekaFeature() {
		return HasFeatures.namedFeature("Eureka Client", EurekaClient.class);
	}


  //导入一个EurekaClientConfigBean类型的bean，这个bean是用来对eureka客户端进行属性配置的
    @Bean
	@ConditionalOnMissingBean(value = EurekaClientConfig.class,
			search = SearchStrategy.CURRENT)
	public EurekaClientConfigBean eurekaClientConfigBean(ConfigurableEnvironment env) {
		return new EurekaClientConfigBean();
	}


    @Bean
	@ConditionalOnMissingBean
	public ManagementMetadataProvider serviceManagementMetadataProvider() {
		return new DefaultManagementMetadataProvider();
	}

    @Bean
	@ConditionalOnMissingBean(value = EurekaInstanceConfig.class,
			search = SearchStrategy.CURRENT)
	public EurekaInstanceConfigBean eurekaInstanceConfigBean(InetUtils inetUtils,
			ManagementMetadataProvider managementMetadataProvider) {
		。。。。。获取属性值
		EurekaInstanceConfigBean instance = new EurekaInstanceConfigBean(inetUtils);
		。。。。。设置属性值
		
		return instance;
	}



//一个eureka服务注册
   @Bean
	public EurekaServiceRegistry eurekaServiceRegistry() {
		return new EurekaServiceRegistry();
	}



//@EnableDiscoveryClient最后也就是为了开启它
	@Bean
	@ConditionalOnBean(AutoServiceRegistrationProperties.class)
	@ConditionalOnProperty(
			value = "spring.cloud.service-registry.auto-registration.enabled",
			matchIfMissing = true)
	public EurekaAutoServiceRegistration eurekaAutoServiceRegistration(
			ApplicationContext context, EurekaServiceRegistry registry,
			EurekaRegistration registration) {
		return new EurekaAutoServiceRegistration(context, registry, registration);
	}


    ///初始化一个 EurekaClient
      @Bean(destroyMethod = "shutdown")
		@ConditionalOnMissingBean(value = EurekaClient.class,
				search = SearchStrategy.CURRENT)
		public EurekaClient eurekaClient(ApplicationInfoManager manager,
				EurekaClientConfig config) {
			return new CloudEurekaClient(manager, config, this.optionalArgs,
					this.context);
		}


//ApplicationInfoManager
		@Bean
		@ConditionalOnMissingBean(value = ApplicationInfoManager.class,
				search = SearchStrategy.CURRENT)
		public ApplicationInfoManager eurekaApplicationInfoManager(
				EurekaInstanceConfig config) {
			InstanceInfo instanceInfo = new InstanceInfoFactory().create(config);
			return new ApplicationInfoManager(config, instanceInfo);
		}


//初始化一个EurekaRegistration
         @Bean
		@ConditionalOnBean(AutoServiceRegistrationProperties.class)
		@ConditionalOnProperty(
				value = "spring.cloud.service-registry.auto-registration.enabled",
				matchIfMissing = true)
		public EurekaRegistration eurekaRegistration(EurekaClient eurekaClient,
				CloudEurekaInstanceConfig instanceConfig,
				ApplicationInfoManager applicationInfoManager, @Autowired(
						required = false) ObjectProvider<HealthCheckHandler> healthCheckHandler) {
			return EurekaRegistration.builder(instanceConfig).with(applicationInfoManager)
					.with(eurekaClient).with(healthCheckHandler).build();
		}



@Bean
		@ConditionalOnMissingBean
		@ConditionalOnEnabledHealthIndicator("eureka")
		public EurekaHealthIndicator eurekaHealthIndicator(EurekaClient eurekaClient,
				EurekaInstanceConfig instanceConfig, EurekaClientConfig clientConfig) {
			return new EurekaHealthIndicator(eurekaClient, instanceConfig, clientConfig);
		}

EurekaClient构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206



 DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
                    Provider<BackupRegistry> backupRegistryProvider, EndpointRandomizer endpointRandomizer) {
     
        if (args != null) {
            this.healthCheckHandlerProvider = args.healthCheckHandlerProvider;
            this.healthCheckCallbackProvider = args.healthCheckCallbackProvider;
            this.eventListeners.addAll(args.getEventListeners());
            this.preRegistrationHandler = args.preRegistrationHandler;
            
        } else {
            this.healthCheckCallbackProvider = null;
            this.healthCheckHandlerProvider = null;
            this.preRegistrationHandler = null;
        }
        健康检查处理提供者，健康检查回调提供者，事件监听，注册信息预处理者的初始化赋值
     ---------------------------------------
     
        
        this.applicationInfoManager = applicationInfoManager;
     	//这个InstanceInfo由EurekaInstanceConfigBean提供
        InstanceInfo myInfo = applicationInfoManager.getInfo();

       //这个config就是EurekaClientConfigBean
        clientConfig = config;
        staticClientConfig = clientConfig;
     
     
       //@NestedConfigurationProperty
	  //private EurekaTransportConfig transport = new CloudEurekaTransportConfig();
     //传输配置
        transportConfig = config.getTransportConfig();
     
     
        instanceInfo = myInfo;
        if (myInfo != null) {
            appPathIdentifier = instanceInfo.getAppName() + "/" + instanceInfo.getId();
        } else {
            logger.warn("Setting instanceInfo to a passed in null value");
        }

     //后备注册服务提供者
        this.backupRegistryProvider = backupRegistryProvider;
     
     //服务地址 随机化者
        this.endpointRandomizer = endpointRandomizer;
     
     
        this.urlRandomizer = new EndpointUtils.InstanceInfoBasedUrlRandomizer(instanceInfo);
     
     //本地范围的apps 的初始化
        localRegionApps.set(new Applications());

     //拉取注册中心的 次数记录
        fetchRegistryGeneration = new AtomicLong(0);

     //去拉取 的远程范围（区域） 初始化
        remoteRegionsToFetch = new AtomicReference<String>(clientConfig.fetchRegistryForRemoteRegions());
     
     //远程区域的引用
        remoteRegionsRef = new AtomicReference<>(remoteRegionsToFetch.get() == null ? null : remoteRegionsToFetch.get().split(","));

        if (config.shouldFetchRegistry()) {
            //是否需要从注册中心拉取，对间隔时间做了一个定义
            this.registryStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRY_PREFIX + "lastUpdateSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L});
        } else {
            this.registryStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
        }

        if (config.shouldRegisterWithEureka()) {
            //是否需要把自己注册到eureka，需要的话也初始化一个时间间隔
            this.heartbeatStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRATION_PREFIX + "lastHeartbeatSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L});
        } else {
            this.heartbeatStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
        }

        logger.info("Initializing Eureka in region {}", clientConfig.getRegion());

        if (!config.shouldRegisterWithEureka() && !config.shouldFetchRegistry()) {
            //如果不需要把自己注册到eureka 并且 不需要去注册中心拉取数据的话，那么一些调度器，心跳执行器，缓存刷新执行器，eureka传输，置为空
            logger.info("Client configured to neither register nor query for data.");
            scheduler = null;
            heartbeatExecutor = null;
            cacheRefreshExecutor = null;
            eurekaTransport = null;
            instanceRegionChecker = new InstanceRegionChecker(new PropertyBasedAzToRegionMapper(config), clientConfig.getRegion());

            // 这是一个允许现有代码使用DiscoveryManager.getInstance（）与DI'd DiscoveryClient合作
            DiscoveryManager.getInstance().setDiscoveryClient(this);
            DiscoveryManager.getInstance().setEurekaClientConfig(config);

            initTimestampMs = System.currentTimeMillis();
            initRegistrySize = this.getApplications().size();
            registrySize = initRegistrySize;
            logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
                    initTimestampMs, initRegistrySize);

            return;  // no need to setup up an network tasks and we are done
        }

        try {
            //heartbeat和cacheRefresh的默认大小为2-1
            //初始一个调度任务线程池
            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

            //初始化传输器，传输器就是拥有 eurekaclient注册，eurekaclient查询，和eurekaclient传输的引用，提供一个公共的shutdown方法
            eurekaTransport = new EurekaTransport();
            
            //调度服务 终端任务，这里只是为了准备好一个eurekaTransport，现在可以传输通信
            scheduleServerEndpointTask(eurekaTransport, args);

            
            
            AzToRegionMapper azToRegionMapper;
            if (clientConfig.shouldUseDnsForFetchingServiceUrls()) {
                //如果是要使用Dns作为拉取 eureka服务器地址，那么这个方法就是把dns
                azToRegionMapper = new DNSBasedAzToRegionMapper(clientConfig);
            } else {
                //否则就使用配置文件的
                azToRegionMapper = new PropertyBasedAzToRegionMapper(clientConfig);
            }
            if (null != remoteRegionsToFetch.get()) {
                //如果配置的从 用于拉取的远程区域，那么加到azToRegionMapper
               azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(","));
            }
            
            instanceRegionChecker = new InstanceRegionChecker(azToRegionMapper, clientConfig.getRegion());
            
            
        } catch (Throwable e) {
            throw new RuntimeException("Failed to initialize DiscoveryClient!", e);
        }

     	//如果需要拉取注册中心数据，则fetchRegistry
        if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) {
            //如果从远程拉取失败，那么从后备实现拉取，一般就是空的
            fetchRegistryFromBackup();
        }
-----------以上---所有的准备工作都做好了--------------------
        // call and execute the pre registration handler before all background tasks (inc registration) is started
     //在启动所有后台任务（inc registration）之前调用并执行预注册处理程序
     
        if (this.preRegistrationHandler != null) {
            //注册信息都拉取完了，开启注册信息预处理
            this.preRegistrationHandler.beforeRegistration();
        }

        if (clientConfig.shouldRegisterWithEureka() && clientConfig.shouldEnforceRegistrationAtInit()) {
            //是否需要把自己注册到eureka服务器上
            //并且是不是再eureka客户端初始化时强制注册，默认是不强制注册，可以指定
            try {
                if (!register() ) {
                    throw new IllegalStateException("Registration error at startup. Invalid server response.");
                }
            } catch (Throwable th) {
                logger.error("Registration error at startup: {}", th.getMessage());
                throw new IllegalStateException(th);
            }
        }

        // 最后，初始化调度任务（例如集群解析器、heartbeat、instanceInfo replicator、fetch
        initScheduledTasks();

        try {
            Monitors.registerObject(this);
        } catch (Throwable e) {
            logger.warn("Cannot register timers", e);
        }

        // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
        // to work with DI'd DiscoveryClient
        DiscoveryManager.getInstance().setDiscoveryClient(this);
        DiscoveryManager.getInstance().setEurekaClientConfig(config);

        initTimestampMs = System.currentTimeMillis();
        initRegistrySize = this.getApplications().size();
        registrySize = initRegistrySize;
        logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
                initTimestampMs, initRegistrySize);
    }

准备传输层的东西

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

private void scheduleServerEndpointTask(EurekaTransport eurekaTransport,
                                           AbstractDiscoveryClientOptionalArgs args) {

           
    //获取附加的过滤器
       Collection<?> additionalFilters = args == null
               ? Collections.emptyList()
               : args.additionalFilters;

    //获取eureka jersey客户端
       EurekaJerseyClient providedJerseyClient = args == null
               ? null
               : args.eurekaJerseyClient;
       
    //获取生产客户端传输对象的 工厂
       TransportClientFactories argsTransportClientFactories = null;
       if (args != null && args.getTransportClientFactories() != null) {
           argsTransportClientFactories = args.getTransportClientFactories();
       }
       
       // Ignore the raw types warnings since the client filter interface changed between jersey 1/2
    //忽略原始类型警告，因为客户端筛选器接口在jersey 1/2之间更改，
       @SuppressWarnings("rawtypes")
       TransportClientFactories transportClientFactories = argsTransportClientFactories == null
               ? new Jersey1TransportClientFactories()
               : argsTransportClientFactories;
               
    //获取Ssl的上下文
       Optional<SSLContext> sslContext = args == null
               ? Optional.empty()
               : args.getSSLContext();
    
    //获取主机名的验证者
       Optional<HostnameVerifier> hostnameVerifier = args == null
               ? Optional.empty()
               : args.getHostnameVerifier();

       // 如果传输工厂没有提供arg，假设他们使用jersey 1作为被动
       eurekaTransport.transportClientFactory = providedJerseyClient == null
               ? transportClientFactories.newTransportClientFactory(clientConfig, additionalFilters, applicationInfoManager.getInfo(), sslContext, hostnameVerifier)
               : transportClientFactories.newTransportClientFactory(additionalFilters, providedJerseyClient);

    //提供一个 applicationSource对象，用于获取本地所有的app
       ApplicationsResolver.ApplicationsSource applicationsSource = new ApplicationsResolver.ApplicationsSource() {
           @Override
           public Applications getApplications(int stalenessThreshold, TimeUnit timeUnit) {
               long thresholdInMs = TimeUnit.MILLISECONDS.convert(stalenessThreshold, timeUnit);
               long delay = getLastSuccessfulRegistryFetchTimePeriod();
               if (delay > thresholdInMs) {
                   logger.info("Local registry is too stale for local lookup. Threshold:{}, actual:{}",
                           thresholdInMs, delay);
                   return null;
               } else {
                   return localRegionApps.get();
               }
           }
       };

    //获取一个辅助对象 主要提高shutdown操作
       eurekaTransport.bootstrapResolver = EurekaHttpClients.newBootstrapResolver(
               clientConfig,
               transportConfig,
               eurekaTransport.transportClientFactory,
               applicationInfoManager.getInfo(),
               applicationsSource,
               endpointRandomizer
       );

       if (clientConfig.shouldRegisterWithEureka()) {
           //判断是否需要注册到eureka服务上，需要就 获取一个专门用于注册的客户端
           
           EurekaHttpClientFactory newRegistrationClientFactory = null;
           EurekaHttpClient newRegistrationClient = null;
           try {
               newRegistrationClientFactory = EurekaHttpClients.registrationClientFactory(
                       eurekaTransport.bootstrapResolver,
                       eurekaTransport.transportClientFactory,
                       transportConfig
               );
               newRegistrationClient = newRegistrationClientFactory.newClient();
           } catch (Exception e) {
               logger.warn("Transport initialization failure", e);
           }
           eurekaTransport.registrationClientFactory = newRegistrationClientFactory;
           eurekaTransport.registrationClient = newRegistrationClient;
       }

       // new method (resolve from primary servers for read) Configure new transport layer (candidate for injecting in the future)
    //新方法（从主服务器解析进行读取）配置新的传输层（将来注入的候选）
       if (clientConfig.shouldFetchRegistry()) {
            //判断是否需要从eureka服务拉取信息，需要就 获取一个专门用于拉取的http客户端
           EurekaHttpClientFactory newQueryClientFactory = null;
           EurekaHttpClient newQueryClient = null;
           try {
               newQueryClientFactory = EurekaHttpClients.queryClientFactory(
                       eurekaTransport.bootstrapResolver,
                       eurekaTransport.transportClientFactory,
                       clientConfig,
                       transportConfig,
                       applicationInfoManager.getInfo(),
                       applicationsSource,
                       endpointRandomizer
               );
               newQueryClient = newQueryClientFactory.newClient();
           } catch (Exception e) {
               logger.warn("Transport initialization failure", e);
           }
           eurekaTransport.queryClientFactory = newQueryClientFactory;
           eurekaTransport.queryClient = newQueryClient;
       }
   }

拉取注册中心

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
private final com.netflix.servo.monitor.Timer FETCH_REGISTRY_TIMER = Monitors
            .newTimer(PREFIX + "FetchRegistry");
 public static Timer newTimer(String name) {
    return newTimer(name, TimeUnit.MILLISECONDS);
  }
  public static Timer newTimer(String name, TimeUnit unit) {
    return new BasicTimer(MonitorConfig.builder(name).build(), unit);
  }





 private boolean fetchRegistry(boolean forceFullRegistryFetch) {
       //启动计时器
        Stopwatch tracer = FETCH_REGISTRY_TIMER.start();

        try {
            // 如果delta（增量）被禁用或是第一次，则获取所有应用程序
            Applications applications = getApplications();

            if (clientConfig.shouldDisableDelta()
                    || (!Strings.isNullOrEmpty(clientConfig.getRegistryRefreshSingleVipAddress()))
                    || forceFullRegistryFetch
                    || (applications == null)
                    || (applications.getRegisteredApplications().size() == 0)
                    || (applications.getVersion() == -1)) //Client application does not have latest library supporting delta
            {
                logger.info("Disable delta property : {}", clientConfig.shouldDisableDelta());
                logger.info("Single vip registry refresh property : {}", clientConfig.getRegistryRefreshSingleVipAddress());
                logger.info("Force full registry fetch : {}", forceFullRegistryFetch);
                logger.info("Application is null : {}", (applications == null));
                logger.info("Registered Applications size is zero : {}",
                        (applications.getRegisteredApplications().size() == 0));
                logger.info("Application version is -1: {}", (applications.getVersion() == -1));
                //从注册拉取全量数据并存储
                getAndStoreFullRegistry();
                
            } else {
                  //从注册拉取增量数据并存储
                getAndUpdateDelta(applications);
            }
            applications.setAppsHashCode(applications.getReconcileHashCode());
            logTotalInstances();
        } catch (Throwable e) {
            logger.error(PREFIX + "{} - was unable to refresh its cache! status = {}", appPathIdentifier, e.getMessage(), e);
            return false;
        } finally {
            if (tracer != null) {
                //计时器暂时，记录一次时间
                tracer.stop();
            }
        }

        // Notify about cache refresh before updating the instance remote status
     //更新实例远程状态之前通知缓存刷新，就是发布一个事件，需要监听器干活
        onCacheRefreshed();

        // Update remote status based on refreshed data held in the cache
     //把本机状态更新，并且发布状态改变事件
        updateInstanceRemoteStatus();

        // registry was fetched successfully, so return true
        return true;
    }
全量数据拉取
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

  private void getAndStoreFullRegistry() throws Throwable {
        long currentUpdateGeneration = fetchRegistryGeneration.get();

        logger.info("Getting all instance registry info from the eureka server");

        Applications apps = null;
      //这里就使用 传输对象的 查询http客户端进行获取
 EurekaHttpResponse<Applications> httpResponse = clientConfig.getRegistryRefreshSingleVipAddress() == null?      eurekaTransport.queryClient.getApplications(remoteRegionsRef.get())
   :eurekaTransport.queryClient.getVip(clientConfig.getRegistryRefreshSingleVipAddress(), remoteRegionsRef.get());
      
      
        if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
            //如果返回的状态码是ok的，就存储
            apps = httpResponse.getEntity();
        }
        logger.info("The response status is {}", httpResponse.getStatusCode());

        if (apps == null) {
            logger.error("The application is null for some reason. Not storing this information");
        } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
            //原子加一，说明只有单个线程进入到这里
            //此时就把拉取到的apps存到localRegionApps
            localRegionApps.set(this.filterAndShuffle(apps));
            logger.debug("Got full registry with apps hashcode {}", apps.getAppsHashCode());
        } else {
            logger.warn("Not updating applications as another thread is updating it already");
        }
    }



/**Gets the <em>applications</em> after filtering the applications for instances with only UP states and shuffling（洗牌，无序处理） them.
     * //在筛选应用程序中只有UP状态的实例并对其进行无序处理后，获取<em>应用程序</em>。
     * 过滤取决于配置{@link EurekaClientConfig#shouldFilterOnlyUpInstances()}指定的选项。
     * The filtering depends on the option specified by the configuration {@link EurekaClientConfig#shouldFilterOnlyUpInstances()}. 
         Shuffling helps in randomizing the applications list there by avoiding the same instances receiving traffic during start ups.
         通过避免相同的实例在启动期间接收流量，Shuffling有助于将应用程序列表随机化。
         */
 private Applications filterAndShuffle(Applications apps) {
        if (apps != null) {
            if (isFetchingRemoteRegionRegistries()) {
                Map<String, Applications> remoteRegionVsApps = new ConcurrentHashMap<String, Applications>();
                apps.shuffleAndIndexInstances(remoteRegionVsApps, clientConfig, instanceRegionChecker);
                for (Applications applications : remoteRegionVsApps.values()) {
                    applications.shuffleInstances(clientConfig.shouldFilterOnlyUpInstances());
                }
                this.remoteRegionVsApps = remoteRegionVsApps;
            } else {
                apps.shuffleInstances(clientConfig.shouldFilterOnlyUpInstances());
            }
        }
        return apps;
    }
获取增量数据
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

private void getAndUpdateDelta(Applications applications) throws Throwable {
       long currentUpdateGeneration = fetchRegistryGeneration.get();

       Applications delta = null;
    
       EurekaHttpResponse<Applications> httpResponse = eurekaTransport.queryClient.getDelta(remoteRegionsRef.get());
    
       if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {
           delta = httpResponse.getEntity();
       }

       if (delta == null) {
           logger.warn("The server does not allow the delta revision to be applied because it is not safe. "
                   + "Hence got the full registry.");
           getAndStoreFullRegistry();
       } else if (fetchRegistryGeneration.compareAndSet(currentUpdateGeneration, currentUpdateGeneration + 1)) {
           logger.debug("Got delta update with apps hashcode {}", delta.getAppsHashCode());
           String reconcileHashCode = "";
           if (fetchRegistryUpdateLock.tryLock()) {
               try {
                   //将从eureka服务器获取的增量信息更新到本地缓存中。
                   updateDelta(delta);
                   //重新获取一次hash值
                   reconcileHashCode = getReconcileHashCode(applications);
               } finally {
                   fetchRegistryUpdateLock.unlock();
               }
           } else {
               logger.warn("Cannot acquire update lock, aborting getAndUpdateDelta");
           }
           // There is a diff in number of instances for some reason
           //如果hash值不同，就代表有改变
           if (!reconcileHashCode.equals(delta.getAppsHashCode()) || clientConfig.shouldLogDeltaDiff()) {
               //协调eureka服务器和客户端注册表信息，并记录差异（如果有）。对账时，应遵循以下流程：
               //远程调用服务器以获取完整的注册表计算和日志差异，原子性设置localregionApps
               reconcileAndLogDifference(delta, reconcileHashCode);  // 这是一个远程呼叫
           }
       } else {
           logger.warn("Not updating application delta as another thread is updating it already");
           logger.debug("Ignoring delta update with apps hashcode {}, as another thread is updating it already", delta.getAppsHashCode());
       }
   }

开启任务调度

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101


private void initScheduledTasks() {
        if (clientConfig.shouldFetchRegistry()) {
            //是否需要拉取注册中心数据
            // registry cache refresh timer
            //获取轮询拉取间隔时间
            int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
           
     // 获取重试延迟的最大乘数值，以防发生一系列超时。就是使用 间隔时间*expBackOffBound就是一个最大延时时间
            int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
            
            cacheRefreshTask = new TimedSupervisorTask(
                    "cacheRefresh",
                    scheduler,
                    cacheRefreshExecutor,
                    registryFetchIntervalSeconds,
                    TimeUnit.SECONDS,
                    expBackOffBound,
                    new CacheRefreshThread()
            );
            //开始调度
            scheduler.schedule(
                    cacheRefreshTask,
                    registryFetchIntervalSeconds, TimeUnit.SECONDS);
        }

    
    
        if (clientConfig.shouldRegisterWithEureka()) {
            //是否注册到eureka服务器
            
            //获取续约间隔
            int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
            
            //更上面的expBackOffBound差不多，就是最大延时时间
            int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
            logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs);

            // Heartbeat timer
            heartbeatTask = new TimedSupervisorTask(
                    "heartbeat",
                    scheduler,
                    heartbeatExecutor,
                    renewalIntervalInSecs,
                    TimeUnit.SECONDS,
                    expBackOffBound,
                    new HeartbeatThread()
            );
            scheduler.schedule(
                    heartbeatTask,
                    renewalIntervalInSecs, TimeUnit.SECONDS);

            
            
            // InstanceInfo replicator
            //InstanceInfo复制器
            
            instanceInfoReplicator = new InstanceInfoReplicator(
                    this,
                    instanceInfo,
                  //获取集群复制的时间间隔
                    clientConfig.getInstanceInfoReplicationIntervalSeconds(),
                    2); // burstSize，突发尺寸

            //初始化一个监听状态变化的监听者
            statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
                @Override
                public String getId() {
                    return "statusChangeListener";
                }

                @Override
                public void notify(StatusChangeEvent statusChangeEvent) {
                    if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||
                            InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) {
                        //如果上一个状态或者现在的状态的 DOWN的，那么警告
                        // log at warn level if DOWN was involved
                        logger.warn("Saw local status change event {}", statusChangeEvent);
                    } else {
                        logger.info("Saw local status change event {}", statusChangeEvent);
                       
                    }
                    
                    //最后按需更新
                    instanceInfoReplicator.onDemandUpdate();
                }
            };

            if (clientConfig.shouldOnDemandUpdateStatusChange()) {
                //注册状态改变监听器，但是必须满足条件
               applicationInfoManager.registerStatusChangeListener(statusChangeListener);
            }
            
            
//间隔多久后，进行 注册或者更新
instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
        } else {
            logger.info("Not registering with Eureka server per configuration");
        }
    }
cacheRefreshTask
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

task=  class CacheRefreshThread implements Runnable {
        public void run() {
            refreshRegistry();
        }
    }
 配置文件：registryFetchIntervalSeconds

timeout=拉取间隔时间
 this.timeoutMillis = timeUnit.toMillis(timeout);   
this.maxDelay = timeoutMillis * expBackOffBound;
 @Override
    public void run() {
        Future<?> future = null;
        try {
            
            //提交任务给线程池
            future = executor.submit(task);
            //记录活动线程数
            threadPoolLevelGauge.set((long) executor.getActiveCount());
            
            //阻塞获取，但是指定了一个超时时间，也就是间隔时间
            future.get(timeoutMillis, TimeUnit.MILLISECONDS);  // block until done or timeout
            //如果指定时间内能获取，则延时时间就是间隔时间
            delay.set(timeoutMillis);
            
            //再次记录活动线程数
            threadPoolLevelGauge.set((long) executor.getActiveCount());
            //成功的计数器++
            successCounter.increment();
        } catch (TimeoutException e) {
            //如果指定时间内未获取
            logger.warn("task supervisor timed out", e);
            //超时计数器++
            timeoutCounter.increment();

            //获取当前延时时间
            long currentDelay = delay.get();
            //当前延时时间*2于最大的延时时间比较，选一个小的，作为下一次延时的时间
            long newDelay = Math.min(maxDelay, currentDelay * 2);
            delay.compareAndSet(currentDelay, newDelay);

        } catch (RejectedExecutionException e) {
            //拒绝，说名线程池关闭了
            if (executor.isShutdown() || scheduler.isShutdown()) {
                logger.warn("task supervisor shutting down, reject the task", e);
            } else {
                logger.warn("task supervisor rejected the task", e);
            }

            rejectedCounter.increment();
        } catch (Throwable e) {
            if (executor.isShutdown() || scheduler.isShutdown()) {
                logger.warn("task supervisor shutting down, can't accept the task");
            } else {
                logger.warn("task supervisor threw an exception", e);
            }

            throwableCounter.increment();
        } finally {
            
            if (future != null) {
                future.cancel(true);
            }

            if (!scheduler.isShutdown()) {
                //重新调度，不过此时的 延时时间可能会长一点
                scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS);
            }
        }
    }




真正执行
     @VisibleForTesting
    void refreshRegistry() {
        try {
            boolean isFetchingRemoteRegionRegistries = isFetchingRemoteRegionRegistries();

            boolean remoteRegionsModified = false;
            // This makes sure that a dynamic change to remote regions to fetch is honored.
            String latestRemoteRegions = clientConfig.fetchRegistryForRemoteRegions();
            if (null != latestRemoteRegions) {
                String currentRemoteRegions = remoteRegionsToFetch.get();
                if (!latestRemoteRegions.equals(currentRemoteRegions)) {
                    // Both remoteRegionsToFetch and AzToRegionMapper.regionsToFetch need to be in sync
                    synchronized (instanceRegionChecker.getAzToRegionMapper()) {
                        if (remoteRegionsToFetch.compareAndSet(currentRemoteRegions, latestRemoteRegions)) {
                            String[] remoteRegions = latestRemoteRegions.split(",");
                            remoteRegionsRef.set(remoteRegions);
                            instanceRegionChecker.getAzToRegionMapper().setRegionsToFetch(remoteRegions);
                            remoteRegionsModified = true;
                        } else {
                            logger.info("Remote regions to fetch modified concurrently," +
                                    " ignoring change from {} to {}", currentRemoteRegions, latestRemoteRegions);
                        }
                    }
                } else {
                    // Just refresh mapping to reflect any DNS/Property change
                    instanceRegionChecker.getAzToRegionMapper().refreshMapping();
                }
            }

            //这里就是拉取注册中心操作，remoteRegionsModified=true为全量拉取，否则增量拉取
            boolean success = fetchRegistry(remoteRegionsModified未true未);
            if (success) {
                //成功了就记录下信息
                registrySize = localRegionApps.get().size();
                lastSuccessfulRegistryFetchTimestamp = System.currentTimeMillis();
            }

            if (logger.isDebugEnabled()) {
                //日志级别是debug就记录下面信息
                StringBuilder allAppsHashCodes = new StringBuilder();
                allAppsHashCodes.append("Local region apps hashcode: ");
                allAppsHashCodes.append(localRegionApps.get().getAppsHashCode());
                allAppsHashCodes.append(", is fetching remote regions? ");
                allAppsHashCodes.append(isFetchingRemoteRegionRegistries);
                for (Map.Entry<String, Applications> entry : remoteRegionVsApps.entrySet()) {
                    allAppsHashCodes.append(", Remote region: ");
                    allAppsHashCodes.append(entry.getKey());
                    allAppsHashCodes.append(" , apps hashcode: ");
                    allAppsHashCodes.append(entry.getValue().getAppsHashCode());
                }
                logger.debug("Completed cache refresh task for discovery. All Apps hash code is {} ",
                        allAppsHashCodes);
            }
        } catch (Throwable e) {
            logger.error("Cannot fetch registry from server", e);
        }
    }
heartbeatTask
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111

task=  private class HeartbeatThread implements Runnable {

        public void run() {
            if (renew()) {
                lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
            }
        }
    }
配置文件：leaseRenewalIntervalInSeconds

timeout=续约间隔时间
 this.timeoutMillis = timeUnit.toMillis(timeout);   
this.maxDelay = timeoutMillis * expBackOffBound;

 @Override
    public void run() {
        Future<?> future = null;
        try {
            future = executor.submit(task);
            threadPoolLevelGauge.set((long) executor.getActiveCount());
            future.get(timeoutMillis, TimeUnit.MILLISECONDS);  // block until done or timeout
            delay.set(timeoutMillis);
            threadPoolLevelGauge.set((long) executor.getActiveCount());
            successCounter.increment();
        } catch (TimeoutException e) {
            logger.warn("task supervisor timed out", e);
            timeoutCounter.increment();

            long currentDelay = delay.get();
            long newDelay = Math.min(maxDelay, currentDelay * 2);
            delay.compareAndSet(currentDelay, newDelay);

        } catch (RejectedExecutionException e) {
            if (executor.isShutdown() || scheduler.isShutdown()) {
                logger.warn("task supervisor shutting down, reject the task", e);
            } else {
                logger.warn("task supervisor rejected the task", e);
            }

            rejectedCounter.increment();
        } catch (Throwable e) {
            if (executor.isShutdown() || scheduler.isShutdown()) {
                logger.warn("task supervisor shutting down, can't accept the task");
            } else {
                logger.warn("task supervisor threw an exception", e);
            }

            throwableCounter.increment();
        } finally {
            if (future != null) {
                future.cancel(true);
            }

            if (!scheduler.isShutdown()) {
                scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS);
            }
        }
    }


真正执行
     boolean renew() {
        EurekaHttpResponse<InstanceInfo> httpResponse;
        try {
            //发送心跳包
            httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
            
            
            logger.debug(PREFIX + "{} - Heartbeat status: {}", appPathIdentifier, httpResponse.getStatusCode());
            
            
            if (httpResponse.getStatusCode() == Status.NOT_FOUND.getStatusCode()) {
                //如果eureka服务器上为找到，则需要注册
                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;
            }
            //最后返回状态与ok比对
            return httpResponse.getStatusCode() == Status.OK.getStatusCode();
        } catch (Throwable e) {
            logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e);
            return false;
        }
    }

注册：

boolean register() throws Throwable {
        logger.info(PREFIX + "{}: registering service...", appPathIdentifier);
        EurekaHttpResponse<Void> httpResponse;
        try {
            //发送注册请求
            httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
        } catch (Exception e) {
            logger.warn(PREFIX + "{} - registration failed {}", appPathIdentifier, e.getMessage(), e);
            throw e;
        }
        if (logger.isInfoEnabled()) {
            logger.info(PREFIX + "{} - registration status: {}", appPathIdentifier, httpResponse.getStatusCode());
        }
    //注册成功，应该返回未连接状态
        return httpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode();
    }
InstanceInfoReplicator
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

 this.replicationIntervalSeconds = replicationIntervalSeconds;
  this.burstSize = 2;//先固定死了，eureka作者还没改
  this.allowedRatePerMinute = 60 * this.burstSize / this.replicationIntervalSeconds;
  this.rateLimiter = new RateLimiter(TimeUnit.MINUTES);
this.discoveryClient = discoveryClient;
  this.instanceInfo = instanceInfo;
this.scheduler = Executors.newScheduledThreadPool(1,
                new ThreadFactoryBuilder()
                        .setNameFormat("DiscoveryClient-InstanceInfoReplicator-%d")
                        .setDaemon(true)
                        .build());
/*


*/
  public boolean onDemandUpdate() {
        if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) {
            if (!scheduler.isShutdown()) {
                scheduler.submit(new Runnable() {
                    @Override
                    public void run() {
                        logger.debug("Executing on-demand update of local InstanceInfo");
    
                        Future latestPeriodic = scheduledPeriodicRef.get();
                        if (latestPeriodic != null && !latestPeriodic.isDone()) {
                            
                            //如果上一次的还没有完成，那么取消执行，就是唤醒所有等待的线程
                            logger.debug("Canceling the latest scheduled update, it will be rescheduled at the end of on demand update");
                            latestPeriodic.cancel(false);
                        }
    
                        InstanceInfoReplicator.this.run();
                    }
                });
                return true;
            } else {
                logger.warn("Ignoring onDemand update due to stopped scheduler");
                return false;
            }
        } else {
            logger.warn("Ignoring onDemand update due to rate limiter");
            return false;
        }
    }



    public void run() {
        try {
            discoveryClient.refreshInstanceInfo();

            Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
            if (dirtyTimestamp != null) {
                discoveryClient.register();
                instanceInfo.unsetIsDirty(dirtyTimestamp);
            }
        } catch (Throwable t) {
            logger.warn("There was a problem with the instance info replicator", t);
        } finally {
            
            //延时调度自己，时间就是  集群复制的间隔时间
            Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
            //存储未来的值
            scheduledPeriodicRef.set(next);
        }
    }


void refreshInstanceInfo() {
        applicationInfoManager.refreshDataCenterInfoIfRequired();
        applicationInfoManager.refreshLeaseInfoIfRequired();

        InstanceStatus status;
        try {
            status = getHealthCheckHandler().getStatus(instanceInfo.getStatus());
        } catch (Exception e) {
            logger.warn("Exception from healthcheckHandler.getStatus, setting status to DOWN", e);
            status = InstanceStatus.DOWN;
        }

        if (null != status) {
            applicationInfoManager.setInstanceStatus(status);
        }
    }

关闭

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
public synchronized void shutdown() {
        if (isShutdown.compareAndSet(false, true)) {
            logger.info("Shutting down DiscoveryClient ...");

            if (statusChangeListener != null && applicationInfoManager != null) {
                applicationInfoManager.unregisterStatusChangeListener(statusChangeListener.getId());
            }

            cancelScheduledTasks();

            // If APPINFO was registered
            if (applicationInfoManager != null
                    && clientConfig.shouldRegisterWithEureka()
                    && clientConfig.shouldUnregisterOnShutdown()) {
                applicationInfoManager.setInstanceStatus(InstanceStatus.DOWN);
                //注销
                unregister();
            }

            if (eurekaTransport != null) {
                //传输层关闭
                eurekaTransport.shutdown();
            }

            heartbeatStalenessMonitor.shutdown();
            registryStalenessMonitor.shutdown();

            //
            Monitors.unregisterObject(this);

            logger.info("Completed shut down of DiscoveryClient");
        }
    }

private void cancelScheduledTasks() {
        if (instanceInfoReplicator != null) {
            instanceInfoReplicator.stop();
        }
        if (heartbeatExecutor != null) {
            heartbeatExecutor.shutdownNow();
        }
        if (cacheRefreshExecutor != null) {
            cacheRefreshExecutor.shutdownNow();
        }
        if (scheduler != null) {
            scheduler.shutdownNow();
        }
        if (cacheRefreshTask != null) {
            cacheRefreshTask.cancel();
        }
        if (heartbeatTask != null) {
            heartbeatTask.cancel();
        }
    }


void unregister() {
        // It can be null if shouldRegisterWithEureka == false
        if(eurekaTransport != null && eurekaTransport.registrationClient != null) {
            try {
                logger.info("Unregistering ...");
                EurekaHttpResponse<Void> httpResponse = eurekaTransport.registrationClient.cancel(instanceInfo.getAppName(), instanceInfo.getId());
                logger.info(PREFIX + "{} - deregister  status: {}", appPathIdentifier, httpResponse.getStatusCode());
            } catch (Exception e) {
                logger.error(PREFIX + "{} - de-registration failed{}", appPathIdentifier, e.getMessage(), e);
            }
        }
    }

配置类

EurekaClientConfig

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
EurekaClientConfig它不集成到cloud的话，使用的是 eureka core 包下的 DefalutEurekaClientConfig

    
集成到cloud就使用EurekaClientConfigBean作为实现
@ConfigurationProperties(EurekaClientConfigBean.PREFIX)
public class EurekaClientConfigBean implements EurekaClientConfig, Ordered {
	public static final String PREFIX = "eureka.client";
    所以配置必须已eureka.client开头，才能配置里面的属性
        
   //拉取数据时间间隔
	private int registryFetchIntervalSeconds = 30;
    //把复制本台机器的实例的变化 给到 eureka服务器 的时间间隔 
	private int instanceInfoReplicationIntervalSeconds = 30;
    //这个时间是延时多久向 eureka服务器 推送自己变化的 时间，没什么用，默认
	private int initialInstanceInfoReplicationIntervalSeconds = 40;
	/**
	 指示轮询eureka服务器信息更改的频率（秒）。可以添加或删除Eureka服务器，此设置控制Eureka客户端应该知道它的时间。 获取服务器地址的时间间隔，因为服务器地址被随机化了（为了流量均衡）
	 */
	private int eurekaServiceUrlPollIntervalSeconds = 5 * MINUTES;
    //代理端口
	private String proxyPort;
    //代理主机
	private String proxyHost;
    //代理名
	private String proxyUserName;
    //密码
	private String proxyPassword;
    //请求的超时时间
	private int eurekaServerReadTimeoutSeconds = 8;
    //此设置会影响实际的连接创建以及从池中获取连接的等待时间。
    //连接的超时时间
	private int eurekaServerConnectTimeoutSeconds = 5;
	//备用注册中心的实现，没有用
	private String backupRegistryImpl;
    //eureka服务器允许的最大连接总数
	private int eurekaServerTotalConnections = 200;
    //per 这个关键头，应该是每一台eureka服务器允许的每一台eureka客户端的最大连接数
	private int eurekaServerTotalConnectionsPerHost = 50;
    //表示eureka注册中心的路径，如果配置为eureka，则为http://x.x.x.x:x/eureka/，在eureka的配置文件中加入此配置表示eureka作为客户端向注册中心注册，从而构成eureka集群。此配置只有在eureka服务器ip地址列表是在DNS中才会用到，默认为null
	private String eurekaServerURLContext;
    
    //获取eureka服务器的端口，此配置只有在eureka服务器ip地址列表是在DNS中才会用到。默认为null
	private String eurekaServerPort;
   
    //获取要查询的DNS名称来获得eureka服务器，此配置只有在eureka服务器ip地址列表是在DNS中才会用到。默认为null
	private String eurekaServerDNSName;

    //eureka客户端是否应该使用DNS机制来获取eureka服务器的地址列表，默认为false
	private boolean useDnsForFetchingServiceUrls = false;
    
    
	//获取实例所在的地区。默认为us-east-1
	private String region = "us-east-1";

	/**
	 * 指示与eureka服务器的HTTP连接在关闭之前可以保持空闲的时间（以秒为单位）。
	 * 在AWS环境中，建议将该值设置为30秒或更短，因为防火墙会在几分钟后清理连接信息，使连接悬而未决。
	 */
    //Eureka服务的http请求关闭之前其响应的时间，默认为30 秒
	private int eurekaConnectionIdleTimeoutSeconds = 30;

	/**
	 * Indicates whether the client is only interested in the registry information for a single VIP.
	 */
    //此客户端只对一个单一的VIP注册表的信息感兴趣。默认为null
	private String registryRefreshSingleVipAddress;

    //心跳执行程序线程池的大小,默认为5
	private int heartbeatExecutorThreadPoolSize = 2;

	//用于计算心跳最大延时时间，10就代表延时最大时间为间隔时间*10
	private int heartbeatExecutorExponentialBackOffBound = 10;

	//执行程序缓存刷新线程池的大小，默认为5
	private int cacheRefreshExecutorThreadPoolSize = 2;

	//用于计算刷新最大延时时间，10就代表延时最大时间为间隔时间*10
	private int cacheRefreshExecutorExponentialBackOffBound = 10;

	
    //  Eureka服务器的连接，默认为http：//XXXX：X/eureka/,但是如果采用DNS方式获取服务地址，则不需要配置此设置。
	private Map<String, String> serviceUrl = new HashMap<>();

	{
		this.serviceUrl.put(DEFAULT_ZONE, DEFAULT_URL);
	}

	//eureka注册表的内容是否被压缩，默认为true，并且是在最好的网络流量下被压缩
	private boolean gZipContent = true;
    
    
    //实例是否在eureka服务器上注册自己的信息以供其他服务发现，默认为true
	private boolean registerWithEureka = true;

    //实例是否使用同一zone里的eureka服务器，默认为true，理想状态下，eureka客户端与服务端是在同一zone下
	private boolean preferSameZoneEureka = true;

    //是否记录eureka服务器和客户端之间在注册表的信息方面的差异，默认为false
	private boolean logDeltaDiff;

	/**
	 * 指示eureka客户端是否应禁用增量数据的获取，而应求助于获取完整的注册表信息。
	 * 请注意，delta获取可以极大地减少流量，因为eureka服务器的更改率通常远低于获取率。
	 * 更改在运行时在registryFetchIntervalSeconds指定的下一个注册表获取周期时生效
	 */
    //默认为false
	private boolean disableDelta;

	/**
	 * Comma separated list of regions for which the eureka registry information will be
	 * fetched. It is mandatory to define the availability zones for each of these regions
	 * as returned by availabilityZones. Failing to do so, will result in failure of
	 * discovery client startup.
	 *
	 */
    //eureka服务注册表信息里的以逗号隔开的地区名单，如果不这样返回这些地区名单，则客户端启动将会出错。默认为null
	private String fetchRemoteRegionsRegistry;

    // 获取实例所在的地区下可用性的区域列表，用逗号隔开
	private Map<String, String> availabilityZones = new HashMap<>();

	//是否获得处于开启状态的实例的应用程序过滤之后的应用程序。默认为true
	private boolean filterOnlyUpInstances = true;

    //此客户端是否获取eureka服务器注册表上的注册信息，默认为true
	private boolean fetchRegistry = true;

    // eureka服务器序列化/反序列化的信息中获取“$”符号的的替换字符串。默认为“_-”
	private String dollarReplacement = "_-";

	//eureka服务器序列化/反序列化的信息中获取“_”符号的的替换字符串。默认为“__”
	private String escapeCharReplacement = "__";

	
    // 服务器是否能够重定向客户端请求到备份服务器。 如果设置为false，服务器将直接处理请求，如果设置为true，它可能发送HTTP重定向到客户端。默认为false
	private boolean allowRedirects = false;

	//如果设置为true,客户端的状态更新将会点播更新到远程服务器上，默认为true
	private boolean onDemandUpdateStatusChange = true;

    //这是一个短暂的编码器的配置，如果最新的编码器是稳定的，则可以去除，默认为null
	private String encoderName;
    
	//这是一个短暂的解码器的配置，如果最新的解码器是稳定的，则可以去除，默认为null
	private String decoderName;

    //客户端数据接收
	private String clientDataAccept = EurekaAccept.full.name();

	/**
	 * 指示客户端是否应在客户端关闭时从远程服务器显式注销自己。
	 */
	private boolean shouldUnregisterOnShutdown = true;

	/**
	 * 指示客户端是否应在初始化期间强制注册。
	 * Defaults to false.
	 */
	private boolean shouldEnforceRegistrationAtInit = false;
}

EurekaInstanceConfig

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147


为集成使用CloudInstanceConfig

    
    集成使用EurekaInstanceConfigBean
@ConfigurationProperties("eureka.instance")    
public class EurekaInstanceConfigBean
		implements CloudEurekaInstanceConfig, EnvironmentAware {
    
    /**
	 * Default prefix for actuator endpoints.
	 */
	private String actuatorPrefix = "/actuator";

	//获得在eureka服务上注册的应用程序的名字，默认为unknow
	private String appname = UNKNOWN;

	// 获得在eureka服务上注册的应用程序组的名字，默认为unknow
	private String appGroupName;

    // 实例注册到eureka服务器时，是否开启通讯，默认为false
	private boolean instanceEnabledOnit;

	//获取该实例应该接收通信的非安全端口。默认为80
	private int nonSecurePort = 80;

	//获取该实例应该接收通信的安全端口，默认为443
	private int securePort = 443;

    //该实例应该接收通信的非安全端口是否启用，默认为true
	private boolean nonSecurePortEnabled = true;
    
	//该实例应该接收通信的安全端口是否启用，默认为false
	private boolean securePortEnabled;

    //eureka客户需要多长时间发送心跳给eureka服务器，表明它仍然活着,默认为30 秒
	private int leaseRenewalIntervalInSeconds = 30;

    //Eureka服务器在接收到实例的最后一次发出的心跳后，需要等待多久才可以将此实例删除，默认为90秒
	private int leaseExpirationDurationInSeconds = 90;

    //此实例定义的虚拟主机名，其他实例将通过使用虚拟主机名找到该实例。
	private String virtualHostName = UNKNOWN;

    //此实例注册到eureka服务端的唯一的实例ID,其组成为${spring.application.name}:${spring.application.instance_id:${random.value}}
	private String instanceId;

    //此实例定义的安全虚拟主机名
	private String secureVirtualHostName = UNKNOWN;

    //与此实例相关联 AWS自动缩放组名称。此项配置是在AWS环境专门使用的实例启动，它已被用于流量停用后自动把一个实例退出服务。
	private String aSGName;

    //获取与此实例相关联的元数据(key,value)。这个信息被发送到eureka服务器，其他实例可以使用。
	private Map<String, String> metadataMap = new HashMap<>();

	/**
	 * Returns the data center this instance is deployed. This information is used to get
	 * some AWS specific instance information if the instance is deployed in AWS.
	 */
    // 该实例被部署在数据中心
	private DataCenterInfo dataCenterInfo = new MyDataCenterInfo(
			DataCenterInfo.Name.MyOwn);

    //获取实例的ip地址
	private String ipAddress;

    //获取此实例状态页的URL路径，然后构造出主机名，安全端口等，默认为/info
	private String statusPageUrlPath = actuatorPrefix + "/info";

    //获取此实例绝对状态页的URL路径，为其他服务提供信息时来找到这个实例的状态的路径，默认为null
	private String statusPageUrl;

    //获取此实例的相关主页URL路径，然后构造出主机名，安全端口等，默认为/
	private String homePageUrlPath = "/";

	/**
	 * Gets the absolute home page URL for this instance. The users can provide the
	 * homePageUrlPath if the home page resides in the same instance talking to eureka,
	 * else in the cases where the instance is a proxy for some other server, users can
	 * provide the full URL. If the full URL is provided it takes precedence.
	 *
	 * It is normally used for informational purposes for other services to use it as a
	 * landing page. The full URL should follow the format http://${eureka.hostname}:7001/
	 * where the value ${eureka.hostname} is replaced at runtime.
	 */
    //获取此实例的绝对主页URL路径，为其他服务提供信息时使用的路径,默认为null
	private String homePageUrl;

	//获取此实例的相对健康检查URL路径，默认为/health
	private String healthCheckUrlPath = actuatorPrefix + "/health";

	/**
	 * Gets the absolute health check page URL for this instance. The users can provide
	 * the healthCheckUrlPath if the health check page resides in the same instance
	 * talking to eureka, else in the cases where the instance is a proxy for some other
	 * server, users can provide the full URL. If the full URL is provided it takes
	 * precedence.
	 *
	 * <p>
	 * It is normally used for making educated decisions based on the health of the
	 * instance - for example, it can be used to determine whether to proceed deployments
	 * to an entire farm or stop the deployments without causing further damage. The full
	 * URL should follow the format http://${eureka.hostname}:7001/ where the value
	 * ${eureka.hostname} is replaced at runtime.
	 */
    //获取此实例的绝对健康检查URL路径,默认为null
	private String healthCheckUrl;

	/**
	 * Gets the absolute secure health check page URL for this instance. The users can
	 * provide the secureHealthCheckUrl if the health check page resides in the same
	 * instance talking to eureka, else in the cases where the instance is a proxy for
	 * some other server, users can provide the full URL. If the full URL is provided it
	 * takes precedence.
	 *
	 * <p>
	 * It is normally used for making educated decisions based on the health of the
	 * instance - for example, it can be used to determine whether to proceed deployments
	 * to an entire farm or stop the deployments without causing further damage. The full
	 * URL should follow the format http://${eureka.hostname}:7001/ where the value
	 * ${eureka.hostname} is replaced at runtime.
	 */
    //获取此实例的绝对安全健康检查网页的URL路径，默认为null
	private String secureHealthCheckUrl;

    //获取用于查找属性的命名空间，默认为eureka
	private String namespace = "eureka";

	/**
	 * 主机名，如果可以在配置时确定（否则将从操作系统 猜测）。
	 */
    // 与此实例相关联的主机名，是其他实例可以用来进行请求的准确名称
	private String hostname;

	/**
	 * 表示在猜测主机名时，服务器的IP地址应在操作系统报告的主机名引用中使用。
	 */
	private boolean preferIpAddress = false;

	/**
	 *向远程Eureka服务器注册的初始状态。
	 */
	private InstanceStatus initialStatus = InstanceStatus.UP;
    
}

服务端

配置类

EurekaServerConfig

没有集成cloud默认实现DefaultEurekaServerConfig

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171



@ConfigurationProperties(EurekaServerConfigBean.PREFIX)
public class EurekaServerConfigBean implements EurekaServerConfig {
 /**
	 * Eureka server configuration properties prefix.
	 */
	public static final String PREFIX = "eureka.server";
    
    private static final int MINUTES = 60 * 1000;

	@Autowired(required = false)
	PropertyResolver propertyResolver;

	private String aWSAccessId;

	private String aWSSecretKey;

	private int eIPBindRebindRetries = 3;

	private int eIPBindingRetryIntervalMs = 5 * MINUTES;

	private int eIPBindingRetryIntervalMsWhenUnbound = 1 * MINUTES;

	private boolean enableSelfPreservation = true;

	private double renewalPercentThreshold = 0.85;

	private int renewalThresholdUpdateIntervalMs = 15 * MINUTES;

	private int peerEurekaNodesUpdateIntervalMs = 10 * MINUTES;

	private int numberOfReplicationRetries = 5;

	private int peerEurekaStatusRefreshTimeIntervalMs = 30 * 1000;

	private int waitTimeInMsWhenSyncEmpty = 5 * MINUTES;

	private int peerNodeConnectTimeoutMs = 200;

	private int peerNodeReadTimeoutMs = 200;

	private int peerNodeTotalConnections = 1000;

	private int peerNodeTotalConnectionsPerHost = 500;

	private int peerNodeConnectionIdleTimeoutSeconds = 30;

	private long retentionTimeInMSInDeltaQueue = 3 * MINUTES;

	private long deltaRetentionTimerIntervalInMs = 30 * 1000;

	private long evictionIntervalTimerInMs = 60 * 1000;

	private int aSGQueryTimeoutMs = 300;

	private long aSGUpdateIntervalMs = 5 * MINUTES;

	private long aSGCacheExpiryTimeoutMs = 10 * MINUTES; // defaults to longer than the

	// asg update interval

	private long responseCacheAutoExpirationInSeconds = 180;

	private long responseCacheUpdateIntervalMs = 30 * 1000;

	private boolean useReadOnlyResponseCache = true;

	private boolean disableDelta;

	private long maxIdleThreadInMinutesAgeForStatusReplication = 10;

	private int minThreadsForStatusReplication = 1;

	private int maxThreadsForStatusReplication = 1;

	private int maxElementsInStatusReplicationPool = 10000;

	private boolean syncWhenTimestampDiffers = true;

	private int registrySyncRetries = 0;

	private long registrySyncRetryWaitMs = 30 * 1000;

	private int maxElementsInPeerReplicationPool = 10000;

	private long maxIdleThreadAgeInMinutesForPeerReplication = 15;

	private int minThreadsForPeerReplication = 5;

	private int maxThreadsForPeerReplication = 20;

	private int maxTimeForReplication = 30000;

	private boolean primeAwsReplicaConnections = true;

	private boolean disableDeltaForRemoteRegions;

	private int remoteRegionConnectTimeoutMs = 1000;

	private int remoteRegionReadTimeoutMs = 1000;

	private int remoteRegionTotalConnections = 1000;

	private int remoteRegionTotalConnectionsPerHost = 500;

	private int remoteRegionConnectionIdleTimeoutSeconds = 30;

	private boolean gZipContentFromRemoteRegion = true;

	private Map<String, String> remoteRegionUrlsWithName = new HashMap<>();

	private String[] remoteRegionUrls;

	private Map<String, Set<String>> remoteRegionAppWhitelist = new HashMap<>();

	private int remoteRegionRegistryFetchInterval = 30;

	private int remoteRegionFetchThreadPoolSize = 20;

	private String remoteRegionTrustStore = "";

	private String remoteRegionTrustStorePassword = "changeit";

	private boolean disableTransparentFallbackToOtherRegion;

	private boolean batchReplication;

	private boolean rateLimiterEnabled = false;

	private boolean rateLimiterThrottleStandardClients = false;

	private Set<String> rateLimiterPrivilegedClients = Collections.emptySet();

	private int rateLimiterBurstSize = 10;

	private int rateLimiterRegistryFetchAverageRate = 500;

	private int rateLimiterFullFetchAverageRate = 100;

	private boolean logIdentityHeaders = true;

	private String listAutoScalingGroupsRoleName = "ListAutoScalingGroups";

	private boolean enableReplicatedRequestCompression = false;

	private String jsonCodecName;

	private String xmlCodecName;

	private int route53BindRebindRetries = 3;

	private int route53BindingRetryIntervalMs = 5 * MINUTES;

	private long route53DomainTTL = 30;

	private AwsBindingStrategy bindingStrategy = AwsBindingStrategy.EIP;

	private int minAvailableInstancesForPeerReplication = -1;

	private int initialCapacityOfResponseCache = 1000;

	private int expectedClientRenewalIntervalSeconds = 30;

	private boolean useAwsAsgApi = true;

	private String myUrl;

    
}

EurekaDashboardProperties

1
2
3
4
5
6
7
8
9
10
11
12
13
14
@ConfigurationProperties("eureka.dashboard")
public class EurekaDashboardProperties {
    /**
	 * The path to the Eureka dashboard (relative to the servlet path). Defaults to "/".
	 */
	private String path = "/";

	/**
	 * Flag to enable the Eureka dashboard. Default true.
	 */
	private boolean enabled = true;

    
}

InstanceRegistryProperties

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

@ConfigurationProperties(PREFIX)
public class InstanceRegistryProperties {
    
   
	/**
	 * Prefix for Eureka instance registry properties.
	 */
	public static final String PREFIX = "eureka.instance.registry";

	/*
	 * Default number of expected client, defaults to 1. Setting
	 * expectedNumberOfClientsSendingRenews to non-zero to ensure that even an isolated
	 * server can adjust its eviction policy to the number of registrations (when it's
	 * zero, even a successful registration won't reset the rate threshold in
	 * InstanceRegistry.register()).
	 */
	@Value("${eureka.server.expectedNumberOfRenewsPerMin:1}") // for backwards
	// compatibility
	private int expectedNumberOfClientsSendingRenews = 1;

	/**
	 * Value used in determining when leases are cancelled, default to 1 for standalone.
	 * Should be set to 0 for peer replicated eurekas
	 */
	@Value("${eureka.server.defaultOpenForTrafficCount:1}") // for backwards compatibility
	private int defaultOpenForTrafficCount = 1;
 
}

@EnableEurekaServer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

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

}
这个注解就导入了一个 EurekaServerMarkerConfiguration类

@Configuration(proxyBeanMethods = false)
public class EurekaServerMarkerConfiguration {

	@Bean
	public Marker eurekaServerMarkerBean() {
		return new Marker();
	}

	class Marker {

	}

}
这个配置类就 只给出一个 Marker的bean
1
2
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
  org.springframework.cloud.netflix.eureka.server.EurekaServerAutoConfiguration

EurekaServerAutoConfiguration

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
@Configuration(proxyBeanMethods = false)
@Import(EurekaServerInitializerConfiguration.class)

//这里就是入口类加注解的原因
@ConditionalOnBean(EurekaServerMarkerConfiguration.Marker.class)

//导入两个配置
@EnableConfigurationProperties({ EurekaDashboardProperties.class,
		InstanceRegistryProperties.class })

//设置一个系统属性 spring.http.encoding.force=false 
@PropertySource("classpath:/eureka/server.properties")
public class EurekaServerAutoConfiguration implements WebMvcConfigurer {
    
    /
    //包含Eureka服务器所需的Jersey资源的包列表。
	private static final String[] EUREKA_PACKAGES = new String[] {
			"com.netflix.discovery", "com.netflix.eureka" };

    //暂时不知道，讲道理只是一个描述类
    @Bean
	public HasFeatures eurekaServerFeature() {
		return HasFeatures.namedFeature("Eureka Server",
				EurekaServerAutoConfiguration.class);
	}
    
    //建一个controller
    @Bean
	@ConditionalOnProperty(prefix = "eureka.dashboard", name = "enabled",
			matchIfMissing = true)
	public EurekaController eurekaController() {
		return new EurekaController(this.applicationInfoManager);
	}

    //服务端的一个编码器
    @Bean
	public ServerCodecs serverCodecs() {
		return new CloudServerCodecs(this.eurekaServerConfig);
	}

    //给一个 额外过滤器容器 
    @Bean
	@ConditionalOnMissingBean
	public ReplicationClientAdditionalFilters replicationClientAdditionalFilters() {
		return new ReplicationClientAdditionalFilters(Collections.emptySet());
	}

    //给一个 有集群意识 的实例注册中心，所有的注册，续约，移除都在这里完成
	@Bean
	public PeerAwareInstanceRegistry peerAwareInstanceRegistry(
			ServerCodecs serverCodecs) {
		this.eurekaClient.getApplications(); // force initialization
		return new InstanceRegistry(this.eurekaServerConfig, this.eurekaClientConfig,
				serverCodecs, this.eurekaClient,
				this.instanceRegistryProperties.getExpectedNumberOfClientsSendingRenews(),
				this.instanceRegistryProperties.getDefaultOpenForTrafficCount());
	}
    
    
    //eureka集群（服务端|注册中心）所有节点，是一个工具类来管理对等节点
    @Bean
	@ConditionalOnMissingBean
	public PeerEurekaNodes peerEurekaNodes(PeerAwareInstanceRegistry registry,
			ServerCodecs serverCodecs,
			ReplicationClientAdditionalFilters replicationClientAdditionalFilters) {
		return new RefreshablePeerEurekaNodes(registry, this.eurekaServerConfig,
				this.eurekaClientConfig, serverCodecs, this.applicationInfoManager,
				replicationClientAdditionalFilters);
	}
    
    //eureka服务器的上下文
    @Bean
	@ConditionalOnMissingBean
	public EurekaServerContext eurekaServerContext(ServerCodecs serverCodecs,
			PeerAwareInstanceRegistry registry, PeerEurekaNodes peerEurekaNodes) {
		return new DefaultEurekaServerContext(this.eurekaServerConfig, serverCodecs,
				registry, peerEurekaNodes, this.applicationInfoManager);
	}
    
    //这个eureka服务器辅助启动类
    @Bean
	public EurekaServerBootstrap eurekaServerBootstrap(PeerAwareInstanceRegistry registry,
			EurekaServerContext serverContext) {
		return new EurekaServerBootstrap(this.applicationInfoManager,
				this.eurekaClientConfig, this.eurekaServerConfig, registry,
				serverContext);
	}
    
    
    
    
    /**
	 * Register the Jersey filter.注册中心的过滤器
	 * @param eurekaJerseyApp an {@link Application} for the filter to be registered
	 * @return a jersey {@link FilterRegistrationBean}
	 */
	@Bean
	public FilterRegistrationBean<?> jerseyFilterRegistration(
			javax.ws.rs.core.Application eurekaJerseyApp) {
		FilterRegistrationBean<Filter> bean = new FilterRegistrationBean<Filter>();
		bean.setFilter(new ServletContainer(eurekaJerseyApp));
		bean.setOrder(Ordered.LOWEST_PRECEDENCE);
		bean.setUrlPatterns(
				Collections.singletonList(EurekaConstants.DEFAULT_PREFIX + "/*"));

		return bean;
	}
    
    
    
    /**
	 * Construct a Jersey {@link javax.ws.rs.core.Application} with all the resources
	 * required by the Eureka server. 
	 * @param environment an {@link Environment} instance to retrieve classpath resources
	 * @param resourceLoader a {@link ResourceLoader} instance to get classloader from
	 * @return created {@link Application} object
	 */
    //里面放着eureka服务器需要的所有资源
	@Bean
	public javax.ws.rs.core.Application jerseyApplication(Environment environment,
			ResourceLoader resourceLoader) {

		ClassPathScanningCandidateComponentProvider provider = new ClassPathScanningCandidateComponentProvider(
				false, environment);

		// Filter to include only classes that have a particular annotation.
		//
		provider.addIncludeFilter(new AnnotationTypeFilter(Path.class));
		provider.addIncludeFilter(new AnnotationTypeFilter(Provider.class));

		// Find classes in Eureka packages (or subpackages)
		//
		Set<Class<?>> classes = new HashSet<>();
		for (String basePackage : EUREKA_PACKAGES) {
			Set<BeanDefinition> beans = provider.findCandidateComponents(basePackage);
			for (BeanDefinition bd : beans) {
				Class<?> cls = ClassUtils.resolveClassName(bd.getBeanClassName(),
						resourceLoader.getClassLoader());
				classes.add(cls);
			}
		}

		// Construct the Jersey ResourceConfig
		Map<String, Object> propsAndFeatures = new HashMap<>();
		propsAndFeatures.put(
				// Skip static content used by the webapp
				ServletContainer.PROPERTY_WEB_PAGE_CONTENT_REGEX,
				EurekaConstants.DEFAULT_PREFIX + "/(fonts|images|css|js)/.*");

		DefaultResourceConfig rc = new DefaultResourceConfig(classes);
		rc.setPropertiesAndFeatures(propsAndFeatures);

		return rc;
	}

    
    @Configuration(proxyBeanMethods = false)
	protected static class EurekaServerConfigBeanConfiguration {

        
        //服务端的配置
		@Bean
		@ConditionalOnMissingBean
		public EurekaServerConfig eurekaServerConfig(EurekaClientConfig clientConfig) {
			EurekaServerConfigBean server = new EurekaServerConfigBean();
			if (clientConfig.shouldRegisterWithEureka()) {
				// Set a sensible default if we are supposed to replicate
                // 默认 5次重试机会
				server.setRegistrySyncRetries(5);
			}
			return server;
		}

	}
    
}

PeerAwareInstanceRegistry构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

/*


*/
public InstanceRegistry(EurekaServerConfig serverConfig,
			EurekaClientConfig clientConfig, ServerCodecs serverCodecs,
			EurekaClient eurekaClient, int expectedNumberOfClientsSendingRenews,
			int defaultOpenForTrafficCount) {
		super(serverConfig, clientConfig, serverCodecs, eurekaClient);

		this.expectedNumberOfClientsSendingRenews = expectedNumberOfClientsSendingRenews;
		this.defaultOpenForTrafficCount = defaultOpenForTrafficCount;
	}
 public PeerAwareInstanceRegistryImpl(
            EurekaServerConfig serverConfig,
            EurekaClientConfig clientConfig,
            ServerCodecs serverCodecs,
            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,
        // then we check the status of a potentially existing lease.
        this.instanceStatusOverrideRule = new FirstMatchWinsCompositeRule(new DownOrStartingRule(),
                new OverrideExistsRule(overriddenInstanceStatusMap), new LeaseExistsRule());
    }


  protected AbstractInstanceRegistry(EurekaServerConfig serverConfig, EurekaClientConfig clientConfig, ServerCodecs serverCodecs) {
        this.serverConfig = serverConfig;
        this.clientConfig = clientConfig;
        this.serverCodecs = serverCodecs;
        this.recentCanceledQueue = new CircularQueue<Pair<Long, String>>(1000);
        this.recentRegisteredQueue = new CircularQueue<Pair<Long, String>>(1000);

        this.renewsLastMin = new MeasuredRate(1000 * 60 * 1);

        this.deltaRetentionTimer.schedule(getDeltaRetentionTask(),
                serverConfig.getDeltaRetentionTimerIntervalInMs(),
                serverConfig.getDeltaRetentionTimerIntervalInMs());
    }


这个方法再eurekaContext初始化时调用
   @Override
    public void init(PeerEurekaNodes peerEurekaNodes) throws Exception {
        this.numberOfReplicationsLastMin.start();
        this.peerEurekaNodes = peerEurekaNodes;
        initializedResponseCache();
        scheduleRenewalThresholdUpdateTask();
        initRemoteRegionRegistry();

        try {
            Monitors.registerObject(this);
        } catch (Throwable e) {
            logger.warn("Cannot register the JMX monitor for the InstanceRegistry :", e);
        }
    }

PeerEurekaNodes构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155

记住这一个生命周期类
/*


*/
	RefreshablePeerEurekaNodes(final PeerAwareInstanceRegistry registry,
				final EurekaServerConfig serverConfig,
				final EurekaClientConfig clientConfig, final ServerCodecs serverCodecs,
				final ApplicationInfoManager applicationInfoManager,
				final ReplicationClientAdditionalFilters replicationClientAdditionalFilters) {
			super(registry, serverConfig, clientConfig, serverCodecs,
					applicationInfoManager);
			this.replicationClientAdditionalFilters = replicationClientAdditionalFilters;
		}

  public PeerEurekaNodes(
            PeerAwareInstanceRegistry registry,
            EurekaServerConfig serverConfig,
            EurekaClientConfig clientConfig,
            ServerCodecs serverCodecs,
            ApplicationInfoManager applicationInfoManager) {
        this.registry = registry;
        this.serverConfig = serverConfig;
        this.clientConfig = clientConfig;
        this.serverCodecs = serverCodecs;
        this.applicationInfoManager = applicationInfoManager;
    }


有servercontext调用
public void start() {
        taskExecutor = Executors.newSingleThreadScheduledExecutor(
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r, "Eureka-PeerNodesUpdater");
                        thread.setDaemon(true);
                        return thread;
                    }
                }
        );
        try {
            //更新集合
            updatePeerEurekaNodes(resolvePeerUrls());
            
            //初始化一个任务，就是用来更新集合
            Runnable peersUpdateTask = new Runnable() {
                @Override
                public void run() {
                    try {
                        updatePeerEurekaNodes(resolvePeerUrls());
                    } catch (Throwable e) {
                        logger.error("Cannot update the replica Nodes", e);
                    }

                }
            };
            //循环调度，间隔时间
            //	private int peerEurekaNodesUpdateIntervalMs = 10 * MINUTES;
            //默认10分钟，这个一半也不会变
            taskExecutor.scheduleWithFixedDelay(
                    peersUpdateTask,
                    serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
                    serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
                    TimeUnit.MILLISECONDS
            );
        } catch (Exception e) {
            throw new IllegalStateException(e);
        }
        for (PeerEurekaNode node : peerEurekaNodes) {
            logger.info("Replica node URL:  {}", node.getServiceUrl());
        }
    }

解析对等节点的url
protected List<String> resolvePeerUrls() {
    //拿到eureka服务器自己自带的客户端实例信息
        InstanceInfo myInfo = applicationInfoManager.getInfo();
    
    //获取一个地区
        String zone = InstanceInfo.getZone(clientConfig.getAvailabilityZones(clientConfig.getRegion()), myInfo);
    
    //这里要么取dns里面取，要么就在本地配置类中取
        List<String> replicaUrls = EndpointUtils
                .getDiscoveryServiceUrls(clientConfig, zone, new EndpointUtils.InstanceInfoBasedUrlRandomizer(myInfo));

    //然后把自己的url过滤掉
        int idx = 0;
        while (idx < replicaUrls.size()) {
            if (isThisMyUrl(replicaUrls.get(idx))) {
                replicaUrls.remove(idx);
            } else {
                idx++;
            }
        }
        return replicaUrls;
    }


---更新对等节点----
    
     protected void updatePeerEurekaNodes(List<String> newPeerUrls) {
        if (newPeerUrls.isEmpty()) {
            logger.warn("The replica size seems to be empty. Check the route 53 DNS Registry");
            return;
        }

    //把原来对等节点存起来
        Set<String> toShutdown = new HashSet<>(peerEurekaNodeUrls);
    //从原来对等集合的移除新进来的
        toShutdown.removeAll(newPeerUrls);
    //把新进来的存起来
        Set<String> toAdd = new HashSet<>(newPeerUrls);
    //新进来的移除原有的
        toAdd.removeAll(peerEurekaNodeUrls);

        if (toShutdown.isEmpty() && toAdd.isEmpty()) { // No change
            都没空，说明没有改变，不需要做逻辑处理
            return;
        }

        // Remove peers no long available
    //还是把原来存起来
        List<PeerEurekaNode> newNodeList = new ArrayList<>(peerEurekaNodes);
//如果原来的不为空
        if (!toShutdown.isEmpty()) {
            logger.info("Removing no longer available peer nodes {}", toShutdown);
            int i = 0;
            while (i < newNodeList.size()) {
                PeerEurekaNode eurekaNode = newNodeList.get(i);
                if (toShutdown.contains(eurekaNode.getServiceUrl())) {
                    移除并关闭
                    newNodeList.remove(i);
                    eurekaNode.shutDown();
                } else {
                    i++;
                }
            }
        }

        // Add new peers
        if (!toAdd.isEmpty()) {
            logger.info("Adding new peer nodes {}", toAdd);
            for (String peerUrl : toAdd) {
                新进来的包装一下加进去
                    
                newNodeList.add(createPeerEurekaNode(peerUrl));
            }
        }

    //最后存一份处理好的
        this.peerEurekaNodes = newNodeList;
        this.peerEurekaNodeUrls = new HashSet<>(newPeerUrls);
    }

对等节点构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39


PeerEurekaNode(PeerAwareInstanceRegistry registry, String targetHost, String serviceUrl,
                                     HttpReplicationClient replicationClient, EurekaServerConfig config,
                                     int batchSize, long maxBatchingDelayMs,
                                     long retrySleepTimeMs, long serverUnavailableSleepTimeMs) {
        this.registry = registry;
        this.targetHost = targetHost;
        this.replicationClient = replicationClient;

        this.serviceUrl = serviceUrl;
        this.config = config;
        this.maxProcessingDelayMs = config.getMaxTimeForReplication();

        String batcherName = getBatcherName();
        ReplicationTaskProcessor taskProcessor = new ReplicationTaskProcessor(targetHost, replicationClient);
    
    创建两个分发器执行，这个分发器只是来控制 任务堆积问题的，就是把任务加到一个 队列
        但是此时并不会执行，需要另外一个执行器去获取任务，然后才能执行
        this.batchingDispatcher = TaskDispatchers.createBatchingTaskDispatcher(
                batcherName,
                config.getMaxElementsInPeerReplicationPool(),
                batchSize,
                config.getMaxThreadsForPeerReplication(),
                maxBatchingDelayMs,
                serverUnavailableSleepTimeMs,
                retrySleepTimeMs,
                taskProcessor
        );
        this.nonBatchingDispatcher = TaskDispatchers.createNonBatchingTaskDispatcher(
                targetHost,
                config.getMaxElementsInStatusReplicationPool(),
                config.getMaxThreadsForStatusReplication(),
                maxBatchingDelayMs,
                serverUnavailableSleepTimeMs,
                retrySleepTimeMs,
                taskProcessor
        );
    }

EurekaServerContext构建

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

/*


*/

 @Inject
    public DefaultEurekaServerContext(EurekaServerConfig serverConfig,
                               ServerCodecs serverCodecs,
                               PeerAwareInstanceRegistry registry,
                               PeerEurekaNodes peerEurekaNodes,
                               ApplicationInfoManager applicationInfoManager) {
        this.serverConfig = serverConfig;
        this.serverCodecs = serverCodecs;
        this.registry = registry;
        this.peerEurekaNodes = peerEurekaNodes;
        this.applicationInfoManager = applicationInfoManager;
    }


//这个注解是jdk的，但是spring检测到这个注解一定会进行执行这个方法，前提是这个类被加入到了springbean定义中
  @PostConstruct
    @Override
    public void initialize() {
        logger.info("Initializing ...");
        peerEurekaNodes.start();//调用了所有对等eureka节点的start()
        try {
            registry.init(peerEurekaNodes);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        logger.info("Initialized");
    }

    @PreDestroy
    @Override
    public void shutdown() {
        logger.info("Shutting down ...");
        registry.shutdown();
        peerEurekaNodes.shutdown();
        ServoControl.shutdown();
        EurekaMonitors.shutdown();
        logger.info("Shut down");
    }

EurekaServerBootstrap构建

1
2
3
4
5
6
7
8
9
10
11
12

没集成使用EurekaBootStrap

	public EurekaServerBootstrap(ApplicationInfoManager applicationInfoManager,
			EurekaClientConfig eurekaClientConfig, EurekaServerConfig eurekaServerConfig,
			PeerAwareInstanceRegistry registry, EurekaServerContext serverContext) {
		this.applicationInfoManager = applicationInfoManager;
		this.eurekaClientConfig = eurekaClientConfig;
		this.eurekaServerConfig = eurekaServerConfig;
		this.registry = registry;
		this.serverContext = serverContext;
	}

@Import(EurekaServerInitializerConfiguration.class)

这个是一个初始化 eureka服务配置的类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

@Configuration(proxyBeanMethods = false)
public class EurekaServerInitializerConfiguration
		implements ServletContextAware, SmartLifecycle, Ordered {
 //他实现了spring的   SmartLifecycle接口，那么在完成所有剩余单例bean注册之后，就会调用start方法
    
    @Override
	public void start() {
		new Thread(() -> {
			try {
				// TODO: is this class even needed now?
                
                //此时利用eurekaServerBootstrap来对上下文初始化，并且把servletContext传过去
				eurekaServerBootstrap.contextInitialized(
						EurekaServerInitializerConfiguration.this.servletContext);
				log.info("Started Eureka Server");

				publish(new EurekaRegistryAvailableEvent(getEurekaServerConfig()));
				EurekaServerInitializerConfiguration.this.running = true;
				publish(new EurekaServerStartedEvent(getEurekaServerConfig()));
			}
			catch (Exception ex) {
				// Help!
				log.error("Could not initialize Eureka servlet context", ex);
			}
		}).start();
	}
    
    
}

bootstrap辅助启动

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
public void contextInitialized(ServletContext context) {
		try {
			initEurekaEnvironment();
			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();
		}

    //填充这个句柄，把serverContext作为操作对象给句柄
		EurekaServerContextHolder.initialize(this.serverContext);

		log.info("Initialized server context");

		// Copy registry from neighboring eureka node
		int registryCount = this.registry.syncUp();
		this.registry.openForTraffic(this.applicationInfoManager, registryCount);

		// Register all monitoring statistics.
		EurekaMonitors.registerAllStats()
            
}



---syncUp----
 @Override
    public int syncUp() {
        // Copy entire entry from neighboring DS node
        int count = 0;

    //getRegistrySyncRetries就是前面设置的5次默认值
        for (int i = 0; ((i < serverConfig.getRegistrySyncRetries()) && (count == 0)); i++) {
            if (i > 0) {
                try {
                    重试等待的时间
                    Thread.sleep(serverConfig.getRegistrySyncRetryWaitMs());
                } catch (InterruptedException e) {
                    logger.warn("Interrupted during registry transfer..");
                    break;
                }
            }
            //因为eureka服务器也有一个 eureka客户端，通过这个辅助的客户端去拉取其他
            //注册中心Applications
            //通过从客户端 获取apps，然后注册自己的 服务端实例注册中心中
            Applications apps = eurekaClient.getApplications();
            for (Application app : apps.getRegisteredApplications()) {
                for (InstanceInfo instance : app.getInstances()) {
                    try {
                        if (isRegisterable(instance)) {
                            register(instance, instance.getLeaseInfo().getDurationInSecs(), true);
                            count++;
                        }
                    } catch (Throwable t) {
                        logger.error("During DS init copy", t);
                    }
                }
            }
        }
        return count;
    }    
    


----如果同步获取一下启动，能得到结果----
    count肯定是有数字的
    如果得不到，就使用默认的
    public void openForTraffic(ApplicationInfoManager applicationInfoManager, int count) {
		super.openForTraffic(applicationInfoManager,
				count == 0 ? this.defaultOpenForTrafficCount : count);
	}


 @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.expectedNumberOfClientsSendingRenews = count;
        //那个期望值的作用就是 设置一个 每次续约最小的阈值
        updateRenewsPerMinThreshold();
        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");
        applicationInfoManager.setInstanceStatus(InstanceStatus.UP);
        
        // 
        super.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());
    }

ServletContainer

1
2
3
4
public class ServletContainer extends HttpServlet implements Filter {

    
 既是一个servelet，有时一个filter

服务端的调度任务

增量数据保留的调度器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// private Timer deltaRetentionTimer = new Timer("Eureka-DeltaRetentionTimer", true);
      //循环调度

  this.deltaRetentionTimer.schedule(getDeltaRetentionTask(),
                serverConfig.getDeltaRetentionTimerIntervalInMs(),
                serverConfig.getDeltaRetentionTimerIntervalInMs());

private TimerTask getDeltaRetentionTask() {
    
    recentlyChangedQueue是一个链表阻塞队列，记录数据变化增量
    
        //这里就是判断消息最后一次的更新时间 与当前系统时间减去 服务端保留消息时间 去比较
        return new TimerTask() {

            @Override
            public void run() {
                Iterator<RecentlyChangedItem> it = recentlyChangedQueue.iterator();
                while (it.hasNext()) {
                    if (it.next().getLastUpdateTime() <
                            System.currentTimeMillis() - serverConfig.getRetentionTimeInMSInDeltaQueue()) {
                        it.remove();
                    } else {
                        break;
                    }
                }
            }

        };
    }

服务剔除任务调度器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

 evictionTaskRef.set(new EvictionTask());
        evictionTimer.schedule(evictionTaskRef.get(),
                serverConfig.getEvictionIntervalTimerInMs(),
                serverConfig.getEvictionIntervalTimerInMs());

class EvictionTask extends TimerTask {

        private final AtomicLong lastExecutionNanosRef = new AtomicLong(0l);

        @Override
        public void run() {
            try {
                long compensationTimeMs = getCompensationTimeMs();
                logger.info("Running the evict task with compensationTime {}ms", compensationTimeMs);
                
                //执行剔除
                evict(compensationTimeMs);
            } catch (Throwable e) {
                logger.error("Could not run the evict task", e);
            }
        }

        /**
         * compute a compensation time defined as the actual time this task was executed since the prev iteration,
         * vs the configured amount of time for execution. This is useful for cases where changes in time (due to
         * clock skew or gc for example) causes the actual eviction task to execute later than the desired time
         * according to the configured cycle.
         */
        long getCompensationTimeMs() {
            long currNanos = getCurrentTimeNano();
            long lastNanos = lastExecutionNanosRef.getAndSet(currNanos);
            if (lastNanos == 0l) {
                return 0l;
            }

            long elapsedMs = TimeUnit.NANOSECONDS.toMillis(currNanos - lastNanos);
            long compensationTime = elapsedMs - serverConfig.getEvictionIntervalTimerInMs();
            return compensationTime <= 0l ? 0l : compensationTime;
        }

        long getCurrentTimeNano() {  // for testing
            return System.nanoTime();
        }

    }

更新对等节点的任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

//初始化一个任务，就是用来更新集合
           Runnable peersUpdateTask = new Runnable() {
               @Override
               public void run() {
                   try {
                       updatePeerEurekaNodes(resolvePeerUrls());
                   } catch (Throwable e) {
                       logger.error("Cannot update the replica Nodes", e);
                   }

               }
           };
           //循环调度，间隔时间
           //	private int peerEurekaNodesUpdateIntervalMs = 10 * MINUTES;
           //默认10分钟，这个一半也不会变
           taskExecutor.scheduleWithFixedDelay(
                   peersUpdateTask,
                   serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
                   serverConfig.getPeerEurekaNodesUpdateIntervalMs(),
                   TimeUnit.MILLISECONDS
           );

续约阈值更新的任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 private void scheduleRenewalThresholdUpdateTask() {
        timer.schedule(new TimerTask() {
                           @Override
                           public void run() {
                               updateRenewalThreshold();
                           }
                       }, serverConfig.getRenewalThresholdUpdateIntervalMs(),
                serverConfig.getRenewalThresholdUpdateIntervalMs());
    }
private void updateRenewalThreshold() {
        try {
            Applications apps = eurekaClient.getApplications();
            int count = 0;
            for (Application app : apps.getRegisteredApplications()) {
                for (InstanceInfo instance : app.getInstances()) {
                    if (this.isRegisterable(instance)) {
                        ++count;
                    }
                }
            }
            synchronized (lock) {
                // Update threshold only if the threshold is greater than the
                // current expected threshold or if self preservation is disabled.
                if ((count) > (serverConfig.getRenewalPercentThreshold() * expectedNumberOfClientsSendingRenews)
                        || (!this.isSelfPreservationModeEnabled())) {
                    this.expectedNumberOfClientsSendingRenews = count;
                    updateRenewsPerMinThreshold();
                }
            }
            logger.info("Current renewal threshold is : {}", numberOfRenewsPerMinThreshold);
        } catch (Throwable e) {
            logger.error("Cannot update renewal threshold", e);
        }
    }

缓存更新任务

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

private TimerTask getCacheUpdateTask() {
     return new TimerTask() {
         @Override
         public void run() {
             logger.debug("Updating the client cache from response cache");
             for (Key key : readOnlyCacheMap.keySet()) {
                 if (logger.isDebugEnabled()) {
                     logger.debug("Updating the client cache from response cache for key : {} {} {} {}",
                             key.getEntityType(), key.getName(), key.getVersion(), key.getType());
                 }
                 try {
                     CurrentRequestVersion.set(key.getVersion());
                     Value cacheValue = readWriteCacheMap.get(key);
                     Value currentCacheValue = readOnlyCacheMap.get(key);
                     if (cacheValue != currentCacheValue) {
                         readOnlyCacheMap.put(key, cacheValue);
                     }
                 } catch (Throwable th) {
                     logger.error("Error while updating the client cache from response cache for key {}", key.toStringCompact(), th);
                 } finally {
                     CurrentRequestVersion.remove();
                 }
             }
         }
     };
 }

服务端怎么收到请求的

ApplicationsResource

1
2
3
4
5
6
7
@Path("/{version}/apps")
@Produces({"application/xml", "application/json"})
public class ApplicationsResource {
    
    

}
1
2
3
4

com.netflix.eureka.resources包下就是接收对应 请求资源的
    
    服务端利用了一个 Response缓存的机制来实现，所以有时候并不会走到 注册中心的注册

ResponseCache

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40


/*

*/


    void invalidate(String appName, @Nullable String vipAddress, @Nullable String secureVipAddress);

    AtomicLong getVersionDelta();

    AtomicLong getVersionDeltaWithRegions();

    /**
     * Get the cached information about applications.
     *
     * <p>
     * If the cached information is not available it is generated on the first
     * request. After the first request, the information is then updated
     * periodically by a background thread.
     * </p>
     *
     * @param key the key for which the cached information needs to be obtained.
     * @return payload which contains information about the applications.
     */
     String get(Key key);

    /**
     * Get the compressed information about the applications.
     *
     * @param key the key for which the compressed cached information needs to be obtained.
     * @return compressed payload which contains information about the applications.
     */
    byte[] getGZIP(Key key);

    /**
     * Performs a shutdown of this cache by stopping internal threads and unregistering
     * Servo monitors.
     */
    void stop();
ResponseCacheImpl
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

 responseCache = new ResponseCacheImpl(serverConfig, serverCodecs, this);


private final java.util.Timer timer = new java.util.Timer("Eureka-CacheFillTimer", true);
    private final AtomicLong versionDelta = new AtomicLong(0);
    private final AtomicLong versionDeltaWithRegions = new AtomicLong(0);

    private final Timer serializeAllAppsTimer = Monitors.newTimer("serialize-all");
    private final Timer serializeDeltaAppsTimer = Monitors.newTimer("serialize-all-delta");
    private final Timer serializeAllAppsWithRemoteRegionTimer = Monitors.newTimer("serialize-all_remote_region");
    private final Timer serializeDeltaAppsWithRemoteRegionTimer = Monitors.newTimer("serialize-all-delta_remote_region");
    private final Timer serializeOneApptimer = Monitors.newTimer("serialize-one");
    private final Timer serializeViptimer = Monitors.newTimer("serialize-one-vip");
    private final Timer compressPayloadTimer = Monitors.newTimer("compress-payload");

    /**
    该映射将没有区域的密钥映射到具有区域（由客户端提供）的密钥列表，因为在无效期间，由本地区域的注册表更改触发，我们不知道客户端请求的区域，我们使用此映射获取所有区域无效的密钥。
     * This map holds mapping of keys without regions to a list of keys with region (provided by clients) Since, during invalidation, triggered by a change in registry for local region, we do not know the regions requested by clients, we use this mapping to get all the keys with regions to be invalidated.
     * If we do not do this, any cached user requests containing region keys will not be invalidated and will stick around till expiry. Github issue: https://github.com/Netflix/eureka/issues/118
     */
    private final Multimap<Key, Key> regionSpecificKeys =
            Multimaps.newListMultimap(new ConcurrentHashMap<Key, Collection<Key>>(), new Supplier<List<Key>>() {
                @Override
                public List<Key> get() {
                    return new CopyOnWriteArrayList<Key>();
                }
            });

//只读缓存
    private final ConcurrentMap<Key, Value> readOnlyCacheMap = new ConcurrentHashMap<Key, Value>();

//读写缓存
    private final LoadingCache<Key, Value> readWriteCacheMap;
//是否使用 ResponseCache;
    private final boolean shouldUseReadOnlyResponseCache;
//注册中心
    private final AbstractInstanceRegistry registry;
//服务端配置
    private final EurekaServerConfig serverConfig;
    private final ServerCodecs serverCodecs;



    ResponseCacheImpl(EurekaServerConfig serverConfig, ServerCodecs serverCodecs, AbstractInstanceRegistry registry) {
        this.serverConfig = serverConfig;
        this.serverCodecs = serverCodecs;
        this.shouldUseReadOnlyResponseCache = serverConfig.shouldUseReadOnlyResponseCache();
        this.registry = registry;

        //获取缓存更新时间间隔
        long responseCacheUpdateIntervalMs = serverConfig.getResponseCacheUpdateIntervalMs();
        //读写缓存，使用CacheBuilder的工厂方法初始化一个指定大小的缓存
        // 还有指定缓存到期时间间隔
        //添加一个缓存移除的一个监听器，主要用于清除regionSpecificKeys
        this.readWriteCacheMap =
                CacheBuilder.newBuilder().initialCapacity(serverConfig.getInitialCapacityOfResponseCache())
                        .expireAfterWrite(serverConfig.getResponseCacheAutoExpirationInSeconds(), TimeUnit.SECONDS)
                        .removalListener(new RemovalListener<Key, Value>() {
                            @Override
                            public void onRemoval(RemovalNotification<Key, Value> notification) {
                                Key removedKey = notification.getKey();
                                if (removedKey.hasRegions()) {
                                    Key cloneWithNoRegions = removedKey.cloneWithoutRegions();
                                    regionSpecificKeys.remove(cloneWithNoRegions, removedKey);
                                }
                            }
                        })
                        .build(new CacheLoader<Key, Value>() {
                            @Override
                            public Value load(Key key) throws Exception {
                                if (key.hasRegions()) {
                                    Key cloneWithNoRegions = key.cloneWithoutRegions();
                                    regionSpecificKeys.put(cloneWithNoRegions, key);
                                }
                                Value value = generatePayload(key);
                                return value;
                            }
                        });

        if (shouldUseReadOnlyResponseCache) {
            如果使用缓存，开一个定时任务
            timer.schedule(getCacheUpdateTask(),
                    new Date(((System.currentTimeMillis() / responseCacheUpdateIntervalMs) * responseCacheUpdateIntervalMs)
                            + responseCacheUpdateIntervalMs),
                    responseCacheUpdateIntervalMs);
        }

        try {
            Monitors.registerObject(this);
        } catch (Throwable e) {
            logger.warn("Cannot register the JMX monitor for the InstanceRegistry", e);
        }
    }
加载指定key的缓存value
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
所以缓存之缓存了Apps和detalApps
  public Value load(Key key) throws Exception {
                                if (key.hasRegions()) {
                                    Key cloneWithNoRegions = key.cloneWithoutRegions();
                                    regionSpecificKeys.put(cloneWithNoRegions, key);
                                }
                                Value value = generatePayload(key);
                                return value;
                            }


private Value generatePayload(Key key) {
        Stopwatch tracer = null;
        try {
            String payload;
            switch (key.getEntityType()) {
                case Application:
                    boolean isRemoteRegionRequested = key.hasRegions();

                    if (ALL_APPS.equals(key.getName())) {
                        if (isRemoteRegionRequested) {
                            tracer = serializeAllAppsWithRemoteRegionTimer.start();
                            payload = getPayLoad(key, registry.getApplicationsFromMultipleRegions(key.getRegions()));
                        } else {
                            tracer = serializeAllAppsTimer.start();
                            //获取所有的Apps
                            payload = getPayLoad(key, registry.getApplications());
                        }
                    } else if (ALL_APPS_DELTA.equals(key.getName())) {
                        if (isRemoteRegionRequested) {
                            tracer = serializeDeltaAppsWithRemoteRegionTimer.start();
                            versionDeltaWithRegions.incrementAndGet();
                            versionDeltaWithRegionsLegacy.incrementAndGet();
                            payload = getPayLoad(key,
                                    registry.getApplicationDeltasFromMultipleRegions(key.getRegions()));
                        } else {
                            tracer = serializeDeltaAppsTimer.start();
                            versionDelta.incrementAndGet();
                            versionDeltaLegacy.incrementAndGet();
                            //获取增量
                            payload = getPayLoad(key, registry.getApplicationDeltas());
                        }
                    } else {
                        tracer = serializeOneApptimer.start();
                        payload = getPayLoad(key, registry.getApplication(key.getName()));
                    }
                    break;
                case VIP:
                case SVIP:
                    tracer = serializeViptimer.start();
                    payload = getPayLoad(key, getApplicationsForVip(key, registry));
                    break;
                default:
                    logger.error("Unidentified entity type: {} found in the cache key.", key.getEntityType());
                    payload = "";
                    break;
            }
            return new Value(payload);
        } finally {
            if (tracer != null) {
                tracer.stop();
            }
        }
    }



  public class Value {
        private final String payload;
        private byte[] gzipped;

        public Value(String payload) {
            this.payload = payload;
            if (!EMPTY_PAYLOAD.equals(payload)) {
                Stopwatch tracer = compressPayloadTimer.start();
                try {
                    ByteArrayOutputStream bos = new ByteArrayOutputStream();
                    GZIPOutputStream out = new GZIPOutputStream(bos);
                    byte[] rawBytes = payload.getBytes();
                    out.write(rawBytes);
                    // Finish creation of gzip file
                    out.finish();
                    out.close();
                    bos.close();
                    gzipped = bos.toByteArray();
                } catch (IOException e) {
                    gzipped = null;
                } finally {
                    if (tracer != null) {
                        tracer.stop();
                    }
                }
            } else {
                gzipped = null;
            }
        }

        public String getPayload() {
            return payload;
        }

        public byte[] getGzipped() {
            return gzipped;
        }

    }
赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia-plus根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

主要涉及技术：
Java后端开发
开源爱好者、Linux

联系QQ:2960571342

很惭愧

只做了一点微小的工作
谢谢大家