资讯专栏INFORMATION COLUMN

SpringCloud(第 050 篇)Netflix Eureka 源码深入剖析(下)

xiaoqibTn / 527人阅读

摘要:于是我们继续断点往下走,发现对象里面只有一个类名路径为。进入看看,这个配置类有哪些重要的方法。。。分析一果不其然,方法被调用了,紧接着也进入断点,然后在往下走,又进入的方法中的回调处。

SpringCloud(第 050 篇)Netflix Eureka 源码深入剖析(下)

-

一、大致介绍

</>复制代码

  1. 1、鉴于一些朋友的提问并提议讲解下eureka的源码分析,由此应运而产生的本章节的内容;
  2. 2、所以我站在自我的理解角度试着整理了这篇Eureka源码的分析,希望对大家有所帮助;
  3. 3、由于篇幅太长不能在一篇里面发布出来,所以拆分了上下篇;
二、基本原理

</>复制代码

  1. 1、Eureka Server 提供服务注册服务,各个节点启动后,会在Eureka Server中进行注册,这样Eureka Server中的服务注册表中将会存储所有可用服务节点的信息,服务节点的信息可以在界面中直观的看到。
  2. 2、Eureka Client 是一个Java 客户端,用于简化与Eureka Server的交互,客户端同时也具备一个内置的、使用轮询负载算法的负载均衡器。
  3. 3、在应用启动后,将会向Eureka Server发送心跳(默认周期为30秒),如果Eureka Server在多个心跳周期没有收到某个节点的心跳,Eureka Server 将会从服务注册表中把这个服务节点移除(默认90秒)。
  4. 4、Eureka Server之间将会通过复制的方式完成数据的同步;
  5. 5、Eureka Client具有缓存的机制,即使所有的Eureka Server 都挂掉的话,客户端依然可以利用缓存中的信息消费其它服务的API;
三、EurekaServer 启动流程分析

详见 SpringCloud(第 049 篇)Netflix Eureka 源码深入剖析(上)

四、EurekaServer 处理服务注册、集群数据复制

详见 SpringCloud(第 049 篇)Netflix Eureka 源码深入剖析(上)

五、EurekaClient 启动流程分析 5.1 调换运行模式,Run运行 springms-discovery-eureka 服务,Debug 运行 springms-provider-user 服务,先观察日志先;

</>复制代码

  1. 2017-10-23 19:43:07.688 INFO 1488 --- [ main] o.s.c.support.DefaultLifecycleProcessor : Starting beans in phase 0
  2. 2017-10-23 19:43:07.694 INFO 1488 --- [ main] o.s.c.n.eureka.InstanceInfoFactory : Setting initial instance status as: STARTING
  3. 2017-10-23 19:43:07.874 INFO 1488 --- [ main] c.n.d.provider.DiscoveryJerseyProvider : Using JSON encoding codec LegacyJacksonJson
  4. 2017-10-23 19:43:07.874 INFO 1488 --- [ main] c.n.d.provider.DiscoveryJerseyProvider : Using JSON decoding codec LegacyJacksonJson
  5. 2017-10-23 19:43:07.971 INFO 1488 --- [ main] c.n.d.provider.DiscoveryJerseyProvider : Using XML encoding codec XStreamXml
  6. 2017-10-23 19:43:07.971 INFO 1488 --- [ main] c.n.d.provider.DiscoveryJerseyProvider : Using XML decoding codec XStreamXml
  7. 2017-10-23 19:43:08.134 INFO 1488 --- [ main] c.n.d.s.r.aws.ConfigClusterResolver : Resolving eureka endpoints via configuration
  8. 2017-10-23 19:43:08.344 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : Disable delta property : false
  9. 2017-10-23 19:43:08.344 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : Single vip registry refresh property : null
  10. 2017-10-23 19:43:08.344 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : Force full registry fetch : false
  11. 2017-10-23 19:43:08.344 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : Application is null : false
  12. 2017-10-23 19:43:08.344 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : Registered Applications size is zero : true
  13. 2017-10-23 19:43:08.344 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : Application version is -1: true
  14. 2017-10-23 19:43:08.345 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : Getting all instance registry info from the eureka server
  15. 2017-10-23 19:43:08.630 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : The response status is 200
  16. 2017-10-23 19:43:08.631 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : Starting heartbeat executor: renew interval is: 30
  17. 2017-10-23 19:43:08.634 INFO 1488 --- [ main] c.n.discovery.InstanceInfoReplicator : InstanceInfoReplicator onDemand update allowed rate per min is 4
  18. 2017-10-23 19:43:08.637 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : Discovery Client initialized at timestamp 1508758988637 with initial instances count: 0
  19. 2017-10-23 19:43:08.657 INFO 1488 --- [ main] c.n.e.EurekaDiscoveryClientConfiguration : Registering application springms-provider-user with eureka with status UP
  20. 2017-10-23 19:43:08.658 INFO 1488 --- [ main] com.netflix.discovery.DiscoveryClient : Saw local status change event StatusChangeEvent [timestamp=1508758988658, current=UP, previous=STARTING]
  21. 2017-10-23 19:43:08.659 INFO 1488 --- [nfoReplicator-0] com.netflix.discovery.DiscoveryClient : DiscoveryClient_SPRINGMS-PROVIDER-USER/springms-provider-user:192.168.3.101:7900: registering service...
  22. 2017-10-23 19:43:08.768 INFO 1488 --- [ main] s.b.c.e.t.TomcatEmbeddedServletContainer : Tomcat started on port(s): 7900 (http)
  23. 2017-10-23 19:43:08.768 INFO 1488 --- [ main] c.n.e.EurekaDiscoveryClientConfiguration : Updating port to 7900
  24. 2017-10-23 19:43:08.773 INFO 1488 --- [ main] c.s.cloud.MsProviderUserApplication : Started MsProviderUserApplication in 9.694 seconds (JVM running for 10.398)
  25. 【【【【【【 用户微服务 】】】】】】已启动.
  26. 【分析一】:根据日志粗粒度看,大多数日志都是在 DiscoveryClient 打印出来的,由此我们先不妨将这些打印日志的地方都打上断点,为了后序
  27. 断点查看调用堆栈信息。
  28. 【分析二】:仔细查看下日志,先是 DefaultLifecycleProcessor 类处理了一些 bean,然后接下来肯定会调用一些实现 SmartLifecycle 类的
  29. start 方法;
  30. 【分析三】: 接着初始化设置了EurekaClient的状态为 STARTING,初始化编码使用的格式,哪些用JSON,哪些用XML;
  31. 【分析四】: 紧接着打印了强制获取注册信息状态为false,已注册的应用大小为0,客户端发送心跳续约,心跳续约间隔为30秒,最后打印Client
  32. 初始化完成;
  33. 【分析五】:带着这些通过日志查看出来的端倪,然后我们还得吸取分析EurekaServer的教训,我们得先去 @EnableEurekaClient 注解瞧瞧。
5.2 有目的性的先去 MsProviderUserApplication 看看,链接点进 EnableEurekaClient 瞧瞧。

</>复制代码

  1. @Target(ElementType.TYPE)
  2. @Retention(RetentionPolicy.RUNTIME)
  3. @Documented
  4. @Inherited
  5. @EnableDiscoveryClient
  6. public @interface EnableEurekaClient {
  7. }
  8. 【分析一】:我们会发现,@EnableEurekaClient 注解类竟然也使用了注解 @EnableDiscoveryClient,那么我们有必要去这个注解类看看。
  9. @Target(ElementType.TYPE)
  10. @Retention(RetentionPolicy.RUNTIME)
  11. @Documented
  12. @Inherited
  13. @Import(EnableDiscoveryClientImportSelector.class)
  14. public @interface EnableDiscoveryClient {
  15. }
  16. 【分析二】:我们看到的是 @EnableDiscoveryClient 注解类有个比较特殊的注解 @Import,由此我们猜想,这里的大多数逻辑是不是都写在这个 EnableDiscoveryClientImportSelector 类呢?
5.3 进入 EnableDiscoveryClientImportSelector 看看到底做了些啥?

</>复制代码

  1. @Order(Ordered.LOWEST_PRECEDENCE - 100)
  2. public class EnableDiscoveryClientImportSelector
  3. extends SpringFactoryImportSelector {
  4. @Override
  5. protected boolean isEnabled() {
  6. return new RelaxedPropertyResolver(getEnvironment()).getProperty(
  7. "spring.cloud.discovery.enabled", Boolean.class, Boolean.TRUE);
  8. }
  9. @Override
  10. protected boolean hasDefaultFactory() {
  11. return true;
  12. }
  13. }
  14. 【分析一】:EnableDiscoveryClientImportSelector 类集成了 SpringFactoryImportSelector 类,但是重写了一个 isEnabled() 方
  15. 法,默认值返回 true,为什么会返回true,也得有个说法吧,于是我们进入父类 EnableDiscoveryClientImportSelector 看看。
  16. /**
  17. * Select and return the names of which class(es) should be imported based on
  18. * the {@link AnnotationMetadata} of the importing @{@link Configuration} class.
  19. */
  20. @Override
  21. public String[] selectImports(AnnotationMetadata metadata) {
  22. if (!isEnabled()) { // 打上断点
  23. return new String[0];
  24. }
  25. AnnotationAttributes attributes = AnnotationAttributes.fromMap(
  26. metadata.getAnnotationAttributes(this.annotationClass.getName(), true));
  27. Assert.notNull(attributes, "No " + getSimpleName() + " attributes found. Is "
  28. + metadata.getClassName() + " annotated with @" + getSimpleName() + "?");
  29. // Find all possible auto configuration classes, filtering duplicates
  30. List factories = new ArrayList<>(new LinkedHashSet<>(SpringFactoriesLoader
  31. .loadFactoryNames(this.annotationClass, this.beanClassLoader)));
  32. if (factories.isEmpty() && !hasDefaultFactory()) {
  33. throw new IllegalStateException("Annotation @" + getSimpleName()
  34. + " found, but there are no implementations. Did you forget to include a starter?");
  35. }
  36. if (factories.size() > 1) {
  37. // there should only ever be one DiscoveryClient, but there might be more than
  38. // one factory
  39. log.warn("More than one implementation " + "of @" + getSimpleName()
  40. + " (now relying on @Conditionals to pick one): " + factories);
  41. }
  42. return factories.toArray(new String[factories.size()]);
  43. }
  44. 【分析二】:发现父类有这么一个 selectImports 方法使用了 isEnabled() 方法,这个方法干了些啥事情呢?我们细看下 selectImports 方法上面的英文注释,大致意思是:选择并且返回需要导入经过注解配置的类,由此我们猜想这个导入的类肯定对我们此次客户端分析有莫大的帮助,于
  45. 是我们现在这个方法打上断点先。于是我们现在该干的事情也干了,没有头绪的时候,我们现在才Run运行EurekaServer,Debug运行springms-provider-user。
5.4 EnableDiscoveryClientImportSelector.selectImports 这个方法果然进断点了。

</>复制代码

  1. 【分析一】:既然进了断点,我们看看这个方法,首先通过注解获取了一些属性,然后加载了一些类名称,于是我们进入 loadFactoryNames 方法看看。
  2. public static List loadFactoryNames(Class factoryClass, ClassLoader classLoader) {
  3. String factoryClassName = factoryClass.getName();
  4. try {
  5. // 注释:public static final String FACTORIES_RESOURCE_LOCATION = "META-INF/spring.factories";
  6. // 注释:这个 jar 包下的一个配置文件
  7. Enumeration urls = (classLoader != null ? classLoader.getResources(FACTORIES_RESOURCE_LOCATION) :
  8. ClassLoader.getSystemResources(FACTORIES_RESOURCE_LOCATION));
  9. List result = new ArrayList();
  10. while (urls.hasMoreElements()) {
  11. URL url = urls.nextElement();
  12. Properties properties = PropertiesLoaderUtils.loadProperties(new UrlResource(url));
  13. String factoryClassNames = properties.getProperty(factoryClassName);
  14. result.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray(factoryClassNames)));
  15. }
  16. return result;
  17. }
  18. catch (IOException ex) {
  19. throw new IllegalArgumentException("Unable to load [" + factoryClass.getName() +
  20. "] factories from location [" + FACTORIES_RESOURCE_LOCATION + "]", ex);
  21. }
  22. }
  23. 【分析二】:加载了一个配置文件,配置文件里面写了啥呢?打开SpringFactoryImportSelector该文件所在的jar包的spring.factories文件一看。
  24. # AutoConfiguration
  25. org.springframework.boot.autoconfigure.EnableAutoConfiguration=
  26. org.springframework.cloud.client.CommonsClientAutoConfiguration,
  27. org.springframework.cloud.client.discovery.noop.NoopDiscoveryClientAutoConfiguration,
  28. org.springframework.cloud.client.hypermedia.CloudHypermediaAutoConfiguration,
  29. org.springframework.cloud.client.loadbalancer.LoadBalancerAutoConfiguration,
  30. org.springframework.cloud.commons.util.UtilAutoConfiguration
  31. # Environment Post Processors
  32. org.springframework.boot.env.EnvironmentPostProcessor=
  33. org.springframework.cloud.client.HostInfoEnvironmentPostProcessor
  34. 【分析三】:看名称,都是一些 Configuration 后缀的类名,所以这些都是加载的一堆堆的配置文件类。于是我们继续断点往下走,发现
  35. factories 对象里面只有一个类名路径为 org.springframework.cloud.netflix.eureka.EurekaDiscoveryClientConfiguration 。看这个
  36. 名字就应该知道这是我们分析EurekaClient的一个重要的配置类,先不管三七二十一,找到该类先。
5.5 进入 EurekaDiscoveryClientConfiguration 看看,这个配置类有哪些重要的方法?

</>复制代码

  1. @Configuration
  2. @EnableConfigurationProperties
  3. @ConditionalOnClass(EurekaClientConfig.class)
  4. @ConditionalOnProperty(value = "eureka.client.enabled", matchIfMissing = true)
  5. @CommonsLog
  6. public class EurekaDiscoveryClientConfiguration implements SmartLifecycle, Ordered {
  7. @Override
  8. public void start() {
  9. // only set the port if the nonSecurePort is 0 and this.port != 0
  10. if (this.port.get() != 0 && this.instanceConfig.getNonSecurePort() == 0) {
  11. this.instanceConfig.setNonSecurePort(this.port.get());
  12. }
  13. // only initialize if nonSecurePort is greater than 0 and it isn"t already running
  14. // because of containerPortInitializer below
  15. if (!this.running.get() && this.instanceConfig.getNonSecurePort() > 0) {
  16. maybeInitializeClient();
  17. if (log.isInfoEnabled()) {
  18. log.info("Registering application " + this.instanceConfig.getAppname()
  19. + " with eureka with status "
  20. + this.instanceConfig.getInitialStatus());
  21. }
  22. this.applicationInfoManager
  23. .setInstanceStatus(this.instanceConfig.getInitialStatus());
  24. if (this.healthCheckHandler != null) {
  25. this.eurekaClient.registerHealthCheck(this.healthCheckHandler);
  26. }
  27. this.context.publishEvent(
  28. new InstanceRegisteredEvent<>(this, this.instanceConfig));
  29. this.running.set(true);
  30. }
  31. }
  32. 。。。 其它省略了
  33. 【分析一】:进入这个类,首先看到该类实现了 SmartLifecycle 接口,那么就肯定会实现 start 方法,而且这个 start 方法感觉应在 “步骤5.1之分析二” 会被加载执行的。
  34. 【分析二】:因为 start 这段代码不多,所以我就索性将 start 方法中的每段代码都点进去看了看,发现 this.applicationInfoManager.setInstanceStatus(this.instanceConfig.getInitialStatus()) 这段代码有一个观察者模式的回调存在。
  35. // ApplicationInfoManager.setInstanceStatus 的方法
  36. public synchronized void setInstanceStatus(InstanceStatus status) {// 打上断点
  37. InstanceStatus prev = instanceInfo.setStatus(status);
  38. if (prev != null) {
  39. for (StatusChangeListener listener : listeners.values()) {
  40. try {
  41. listener.notify(new StatusChangeEvent(prev, status));
  42. } catch (Exception e) {
  43. logger.warn("failed to notify listener: {}", listener.getId(), e);
  44. }
  45. }
  46. }
  47. }
  48. 【分析三】:这个方法会因为状态的改变而回调所有实现 StatusChangeListener 这个类的地方,前提得先注册到 listeners 中去才行。
  49. 【分析四】:于是乎,我们断定,若想要回调,那么就必须有地方先注册这个事件,而且这个注册还必须提前执行在 start 方法前执行,于是我们得先
  50. 在 ApplicationInfoManager 这个类中找到注册到 listeners 的这个方法。
  51. public void registerStatusChangeListener(StatusChangeListener listener) {// 打上断点
  52. listeners.put(listener.getId(), listener);
  53. }
  54. 【分析五】:没错,就是这个方法,肯定有地方调用这个方法,不然的话,那调用 setInstanceStatus 这个方法的意义就什么用了。于是我们逆向找
  55. 下 registerStatusChangeListener 被调用的地方。
  56. 【分析六】:很不巧的是,尽然只有1个地方被调用,这个地方就是 DiscoveryClient.initScheduledTasks 方法,而且 initScheduledTasks
  57. 方法又是在 DiscoveryClient 的构造函数里面调用的,同时我们也对 initScheduledTasks 以及 initScheduledTasks 被调用的构造方法地方
  58. 打上断点。
5.6 由于翻阅代码时间有点久了,因此我们关闭 springms-provider-user 微服务,重新 Debug 运行一下。

</>复制代码

  1. 【分析一】:果不其然,EurekaDiscoveryClientConfiguration.start 方法被调用了,紧接着 this.applicationInfoManager.setInstanceStatus(this.instanceConfig.getInitialStatus()) 也进入断点,然后在往下走,又进入的
  2. DiscoveryClient.initScheduledTasks 方法中的 notify 回调处。
  3. 【分析二】:看着断点依次经过我们上述分析的地方,然后也符合日志打印的顺序,所以我们现在应该是有必要好好看看 DiscoveryClient.initScheduledTasks 这个方法究竟干了什么伟大的事情。然而又想了想,还不如看看 initScheduledTasks 被调用的构造方法。
5.7 进入 DiscoveryClient 经过 @Inject 注解过的构造方法。

</>复制代码

  1. @Inject
  2. DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, DiscoveryClientOptionalArgs args, Provider backupRegistryProvider) {
  3. if (args != null) {
  4. this.healthCheckHandlerProvider = args.healthCheckHandlerProvider;
  5. this.healthCheckCallbackProvider = args.healthCheckCallbackProvider;
  6. this.eventListeners.addAll(args.getEventListeners());
  7. } else {
  8. this.healthCheckCallbackProvider = null;
  9. this.healthCheckHandlerProvider = null;
  10. }
  11. this.applicationInfoManager = applicationInfoManager;
  12. InstanceInfo myInfo = applicationInfoManager.getInfo();
  13. clientConfig = config;
  14. staticClientConfig = clientConfig;
  15. transportConfig = config.getTransportConfig();
  16. instanceInfo = myInfo;
  17. if (myInfo != null) {
  18. appPathIdentifier = instanceInfo.getAppName() + "/" + instanceInfo.getId();
  19. } else {
  20. logger.warn("Setting instanceInfo to a passed in null value");
  21. }
  22. this.backupRegistryProvider = backupRegistryProvider;
  23. this.urlRandomizer = new EndpointUtils.InstanceInfoBasedUrlRandomizer(instanceInfo);
  24. localRegionApps.set(new Applications());
  25. fetchRegistryGeneration = new AtomicLong(0);
  26. remoteRegionsToFetch = new AtomicReference(clientConfig.fetchRegistryForRemoteRegions());
  27. remoteRegionsRef = new AtomicReference<>(remoteRegionsToFetch.get() == null ? null : remoteRegionsToFetch.get().split(","));
  28. if (config.shouldFetchRegistry()) {
  29. this.registryStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRY_PREFIX + "lastUpdateSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L});
  30. } else {
  31. this.registryStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
  32. }
  33. if (config.shouldRegisterWithEureka()) {
  34. this.heartbeatStalenessMonitor = new ThresholdLevelsMetric(this, METRIC_REGISTRATION_PREFIX + "lastHeartbeatSec_", new long[]{15L, 30L, 60L, 120L, 240L, 480L});
  35. } else {
  36. this.heartbeatStalenessMonitor = ThresholdLevelsMetric.NO_OP_METRIC;
  37. }
  38. if (!config.shouldRegisterWithEureka() && !config.shouldFetchRegistry()) {
  39. logger.info("Client configured to neither register nor query for data.");
  40. scheduler = null;
  41. heartbeatExecutor = null;
  42. cacheRefreshExecutor = null;
  43. eurekaTransport = null;
  44. instanceRegionChecker = new InstanceRegionChecker(new PropertyBasedAzToRegionMapper(config), clientConfig.getRegion());
  45. // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
  46. // to work with DI"d DiscoveryClient
  47. DiscoveryManager.getInstance().setDiscoveryClient(this);
  48. DiscoveryManager.getInstance().setEurekaClientConfig(config);
  49. initTimestampMs = System.currentTimeMillis();
  50. logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
  51. initTimestampMs, this.getApplications().size());
  52. return; // no need to setup up an network tasks and we are done
  53. }
  54. try {
  55. // 注释:定时任务调度准备
  56. scheduler = Executors.newScheduledThreadPool(3,
  57. new ThreadFactoryBuilder()
  58. .setNameFormat("DiscoveryClient-%d")
  59. .setDaemon(true)
  60. .build());
  61. // 注释:实例化心跳定时任务线程池
  62. heartbeatExecutor = new ThreadPoolExecutor(
  63. 1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
  64. new SynchronousQueue(),
  65. new ThreadFactoryBuilder()
  66. .setNameFormat("DiscoveryClient-HeartbeatExecutor-%d")
  67. .setDaemon(true)
  68. .build()
  69. ); // use direct handoff
  70. // 注释:实例化缓存刷新定时任务线程池
  71. cacheRefreshExecutor = new ThreadPoolExecutor(
  72. 1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
  73. new SynchronousQueue(),
  74. new ThreadFactoryBuilder()
  75. .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
  76. .setDaemon(true)
  77. .build()
  78. ); // use direct handoff
  79. eurekaTransport = new EurekaTransport();
  80. scheduleServerEndpointTask(eurekaTransport, args);
  81. AzToRegionMapper azToRegionMapper;
  82. if (clientConfig.shouldUseDnsForFetchingServiceUrls()) {
  83. azToRegionMapper = new DNSBasedAzToRegionMapper(clientConfig);
  84. } else {
  85. azToRegionMapper = new PropertyBasedAzToRegionMapper(clientConfig);
  86. }
  87. if (null != remoteRegionsToFetch.get()) {
  88. azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(","));
  89. }
  90. instanceRegionChecker = new InstanceRegionChecker(azToRegionMapper, clientConfig.getRegion());
  91. } catch (Throwable e) {
  92. throw new RuntimeException("Failed to initialize DiscoveryClient!", e);
  93. }
  94. if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) {
  95. fetchRegistryFromBackup();
  96. }
  97. // 注释:初始化调度任务
  98. initScheduledTasks();
  99. try {
  100. Monitors.registerObject(this);
  101. } catch (Throwable e) {
  102. logger.warn("Cannot register timers", e);
  103. }
  104. // This is a bit of hack to allow for existing code using DiscoveryManager.getInstance()
  105. // to work with DI"d DiscoveryClient
  106. DiscoveryManager.getInstance().setDiscoveryClient(this);
  107. DiscoveryManager.getInstance().setEurekaClientConfig(config);
  108. initTimestampMs = System.currentTimeMillis();
  109. logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
  110. initTimestampMs, this.getApplications().size());
  111. }
  112. 【分析一】:从往下看,initScheduledTasks 这个方法顾名思义就是初始化调度任务,所以这里面的内容应该就是重头戏,进入看看。
  113. private void initScheduledTasks() {
  114. if (clientConfig.shouldFetchRegistry()) {
  115. // registry cache refresh timer
  116. // 注释:间隔多久去拉取服务注册信息,默认时间 30秒
  117. int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
  118. int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
  119. // 注释:定时任务,每间隔 30秒 去拉取一次服务注册信息
  120. scheduler.schedule(
  121. new TimedSupervisorTask(
  122. "cacheRefresh",
  123. scheduler,
  124. cacheRefreshExecutor,
  125. registryFetchIntervalSeconds,
  126. TimeUnit.SECONDS,
  127. expBackOffBound,
  128. new CacheRefreshThread()
  129. ),
  130. registryFetchIntervalSeconds, TimeUnit.SECONDS);
  131. }
  132. if (clientConfig.shouldRegisterWithEureka()) {
  133. // 注释:间隔多久发送一次心跳续约,默认间隔时间 30 秒
  134. int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
  135. int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
  136. logger.info("Starting heartbeat executor: " + "renew interval is: " + renewalIntervalInSecs);
  137. // Heartbeat timer
  138. // 注释:定时任务,每间隔 30秒 去想 EurekaServer 发送一次心跳续约
  139. scheduler.schedule(
  140. new TimedSupervisorTask(
  141. "heartbeat",
  142. scheduler,
  143. heartbeatExecutor,
  144. renewalIntervalInSecs,
  145. TimeUnit.SECONDS,
  146. expBackOffBound,
  147. new HeartbeatThread()
  148. ),
  149. renewalIntervalInSecs, TimeUnit.SECONDS);
  150. // InstanceInfo replicator
  151. // 注释:实例信息复制器,定时刷新dataCenterInfo数据中心信息,默认30秒
  152. instanceInfoReplicator = new InstanceInfoReplicator(
  153. this,
  154. instanceInfo,
  155. clientConfig.getInstanceInfoReplicationIntervalSeconds(),
  156. 2); // burstSize
  157. // 注释:实例化状态变化监听器
  158. statusChangeListener = new ApplicationInfoManager.StatusChangeListener() {
  159. @Override
  160. public String getId() {
  161. return "statusChangeListener";
  162. }
  163. @Override
  164. public void notify(StatusChangeEvent statusChangeEvent) {
  165. if (InstanceStatus.DOWN == statusChangeEvent.getStatus() ||
  166. InstanceStatus.DOWN == statusChangeEvent.getPreviousStatus()) {
  167. // log at warn level if DOWN was involved
  168. logger.warn("Saw local status change event {}", statusChangeEvent);
  169. } else {
  170. logger.info("Saw local status change event {}", statusChangeEvent);
  171. }
  172. // 注释:状态有变化的话,会回调这个方法
  173. instanceInfoReplicator.onDemandUpdate();
  174. }
  175. };
  176. // 注释:注册状态变化监听器
  177. if (clientConfig.shouldOnDemandUpdateStatusChange()) {
  178. applicationInfoManager.registerStatusChangeListener(statusChangeListener);
  179. }
  180. instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
  181. } else {
  182. logger.info("Not registering with Eureka server per configuration");
  183. }
  184. }
  185. 【分析二】:在这个方法从上往下一路注释分析下来,干了EurekaClient我们最想知道的一些事情,定时任务获取注册信息,定时任务刷新缓存,定时
  186. 任务心跳续约,定时任务同步数据中心数据,状态变化监听回调等。但是唯独没看到注册,这是怎么回事呢?
  187. 【分析三】:我们忘记了一个重要的方法,instanceInfoReplicator.onDemandUpdate() 就是在状态改变的时候,我们是如何处理的?由此,我们觉得这里面肯定有猫腻,不然没办法注册呀。
  188. public boolean onDemandUpdate() {
  189. if (rateLimiter.acquire(burstSize, allowedRatePerMinute)) {
  190. scheduler.submit(new Runnable() {
  191. @Override
  192. public void run() {
  193. logger.debug("Executing on-demand update of local InstanceInfo");
  194. Future latestPeriodic = scheduledPeriodicRef.get();
  195. if (latestPeriodic != null && !latestPeriodic.isDone()) {
  196. logger.debug("Canceling the latest scheduled update, it will be rescheduled at the end of on demand update");
  197. latestPeriodic.cancel(false);
  198. }
  199. // 注释:这里进行了实例信息刷新和注册
  200. InstanceInfoReplicator.this.run();
  201. }
  202. });
  203. return true;
  204. } else {
  205. logger.warn("Ignoring onDemand update due to rate limiter");
  206. return false;
  207. }
  208. }
  209. 【分析四】:onDemandUpdate 这个方法,看来看去,唯独 InstanceInfoReplicator.this.run() 这个方法还有点用,而且还是 run 方法呢,感情 InstanceInfoReplicator 这个类还是实现了 Runnable 接口?经过查看这个类,还真是实现了 Runnable 接口。
  210. 【分析五】:于是乎,我们有理由相信,这个方法应该我们要找的注册所在的地方,翻开代码看看究竟。
  211. public void run() {
  212. try {
  213. discoveryClient.refreshInstanceInfo();
  214. Long dirtyTimestamp = instanceInfo.isDirtyWithTime();
  215. if (dirtyTimestamp != null) {
  216. discoveryClient.register();
  217. instanceInfo.unsetIsDirty(dirtyTimestamp);
  218. }
  219. } catch (Throwable t) {
  220. logger.warn("There was a problem with the instance info replicator", t);
  221. } finally {
  222. Future next = scheduler.schedule(this, replicationIntervalSeconds, TimeUnit.SECONDS);
  223. scheduledPeriodicRef.set(next);
  224. }
  225. }
  226. 【分析六】:映入眼帘的就是 discoveryClient.register() 这个刺眼的 register 方法,终于有点苗头了,原来注册方法找的这么千辛万苦。虽然找到了这里,但是我还是想看看这个让我们找的千辛万苦的注册方法到底是怎么注册的呢?
  227. boolean register() throws Throwable {
  228. logger.info(PREFIX + appPathIdentifier + ": registering service...");
  229. EurekaHttpResponse httpResponse;
  230. try {
  231. httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
  232. } catch (Exception e) {
  233. logger.warn("{} - registration failed {}", PREFIX + appPathIdentifier, e.getMessage(), e);
  234. throw e;
  235. }
  236. if (logger.isInfoEnabled()) {
  237. logger.info("{} - registration status: {}", PREFIX + appPathIdentifier, httpResponse.getStatusCode());
  238. }
  239. return httpResponse.getStatusCode() == 204;
  240. }
  241. 【分析七】:原来调用了 EurekaHttpClient 封装的客户端请求对象来进行注册的,再继续深探 registrationClient.register 方法,于是我们来到了 AbstractJerseyEurekaHttpClient.register 方法。
  242. @Override
  243. public EurekaHttpResponse register(InstanceInfo info) {
  244. String urlPath = "apps/" + info.getAppName();
  245. ClientResponse response = null;
  246. try {
  247. Builder resourceBuilder = jerseyClient.resource(serviceUrl).path(urlPath).getRequestBuilder();
  248. addExtraHeaders(resourceBuilder);
  249. response = resourceBuilder
  250. .header("Accept-Encoding", "gzip")
  251. .type(MediaType.APPLICATION_JSON_TYPE)
  252. .accept(MediaType.APPLICATION_JSON)
  253. // 注释:打包带上当前应用的所有信息 info
  254. .post(ClientResponse.class, info);
  255. return anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
  256. } finally {
  257. if (logger.isDebugEnabled()) {
  258. logger.debug("Jersey HTTP POST {}/{} with instance {}; statusCode={}", serviceUrl, urlPath, info.getId(),
  259. response == null ? "N/A" : response.getStatus());
  260. }
  261. if (response != null) {
  262. response.close();
  263. }
  264. }
  265. }
  266. 【分析八】:原来调用的是 Jersey RESTful 框架来进行请求的,然后在 EurekaServer 那边就会在 ApplicationResource.addInstance 方法接收客户端的注册请求,因此我们的 EurekaClient 是如何注册的就到此为止了。
  267. 【分析九】:至于那些续约、心跳的流程分析和这个注册的流程大体差不多,相信大家按照我刚刚这么分析断点下去,一定能分析的很到位的。
六、下载地址

https://gitee.com/ylimhhmily/SpringCloudTutorial.git

SpringCloudTutorial交流QQ群: 235322432

SpringCloudTutorial交流微信群: 微信沟通群二维码图片链接

欢迎关注,您的肯定是对我最大的支持!!!

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/67867.html

相关文章

  • SpringCloud 049 Netflix Eureka 源码深入剖析(上)

    摘要:在应用启动后,将会向发送心跳默认周期为秒,如果在多个心跳周期没有收到某个节点的心跳,将会从服务注册表中把这个服务节点移除默认秒。进入类看看,看这个类的名字,见名知意,应该就是的启动类了。。。分析一由于是我们刚刚打断点 SpringCloud(第 049 篇)Netflix Eureka 源码深入剖析(上) - 一、大致介绍 1、鉴于一些朋友的提问并提议讲解下eureka的源码分析,由此...

    niuxiaowei111 评论0 收藏0
  • SpringCloud 021 )Zuul 的过滤器 ZuulFilter 的使用

    摘要:第篇的过滤器的使用一大致介绍我们在学的时候,就有过滤器和拦截器的使用,而同样也有过滤器的使用,本章节我们指在如何简单使用。是否执行该过滤器。说明需要过滤说明不要过滤过滤器的具体逻辑。请求的添加服务网关微服务启动类的过滤器的使用。 SpringCloud(第 021 篇)Zuul 的过滤器 ZuulFilter 的使用 - 一、大致介绍 1、我们在学 Spring 的时候,就有过滤器和拦...

    kumfo 评论0 收藏0
  • SpringCloud 024 )简单文件上传微服务加入zuul微服务后用zuul微服务地址实

    摘要:提供给文件上传微服务用的。注意注解能注册到服务上,是因为该注解包含了客户端的注解,该是一个复合注解。地址可以查看该微服务网关代理了多少微服务的。 SpringCloud(第 024 篇)简单文件上传微服务,并加入 zuul 微服务后用 zuul 微服务地址采取curl或者页面点击实现文件上传 - 一、大致介绍 1、本章节主要将文件上传微服务加入到 zuul 服务中去,然后利用 zuul...

    Cympros 评论0 收藏0

发表评论

0条评论

xiaoqibTn

|高级讲师

TA的文章

阅读更多
最新活动
阅读需要支付1元查看
<