Spring Cloud Eureka源码解析之Eureka Client启动流程（二）

爱被打了一巴掌 2022-09-16 04:52 224阅读 0赞

看下EurekaClient的自动装配类EurekaClientAutoConfiguration  
EurekaClientConfigBean包含了`eureka.client`的配置

@Bean
    @ConditionalOnMissingBean(value = EurekaClientConfig.class, search = SearchStrategy.CURRENT)
    public EurekaClientConfigBean eurekaClientConfigBean(ConfigurableEnvironment env) { 
        EurekaClientConfigBean client = new EurekaClientConfigBean();
        if ("bootstrap".equals(this.env.getProperty("spring.config.name"))) { 
            // We don't register during bootstrap by default, but there will be another
            // chance later.
            client.setRegisterWithEureka(false);
        }
        return client;
    }

EurekaInstanceConfigBean包含了`eureka.instance`的配置

@Bean
    @ConditionalOnMissingBean(value = EurekaInstanceConfig.class, search = SearchStrategy.CURRENT)
    public EurekaInstanceConfigBean eurekaInstanceConfigBean(InetUtils inetUtils,
                                                             ManagementMetadataProvider managementMetadataProvider) { 
        String hostname = getProperty("eureka.instance.hostname");
        boolean preferIpAddress = Boolean.parseBoolean(getProperty("eureka.instance.prefer-ip-address"));
        String ipAddress = getProperty("eureka.instance.ip-address");
        boolean isSecurePortEnabled = Boolean.parseBoolean(getProperty("eureka.instance.secure-port-enabled"));
    
        String serverContextPath = env.getProperty("server.context-path", "/");
        int serverPort = Integer.valueOf(env.getProperty("server.port", env.getProperty("port", "8080")));
    
        Integer managementPort = env.getProperty("management.server.port", Integer.class);// nullable. should be wrapped into optional
        String managementContextPath = env.getProperty("management.server.context-path");// nullable. should be wrapped into optional
        Integer jmxPort = env.getProperty("com.sun.management.jmxremote.port", Integer.class);//nullable
        EurekaInstanceConfigBean instance = new EurekaInstanceConfigBean(inetUtils);
    
        instance.setNonSecurePort(serverPort);
        instance.setInstanceId(getDefaultInstanceId(env));
        instance.setPreferIpAddress(preferIpAddress);
        instance.setSecurePortEnabled(isSecurePortEnabled);
        if (StringUtils.hasText(ipAddress)) { 
            instance.setIpAddress(ipAddress);
        }
    
        if(isSecurePortEnabled) { 
            instance.setSecurePort(serverPort);
        }
    
        if (StringUtils.hasText(hostname)) { 
            instance.setHostname(hostname);
        }
        String statusPageUrlPath = getProperty("eureka.instance.status-page-url-path");
        String healthCheckUrlPath = getProperty("eureka.instance.health-check-url-path");
    
        if (StringUtils.hasText(statusPageUrlPath)) { 
            instance.setStatusPageUrlPath(statusPageUrlPath);
        }
        if (StringUtils.hasText(healthCheckUrlPath)) { 
            instance.setHealthCheckUrlPath(healthCheckUrlPath);
        }
    
        ManagementMetadata metadata = managementMetadataProvider.get(instance, serverPort,
                                                                     serverContextPath, managementContextPath, managementPort);
    
        if(metadata != null) { 
            instance.setStatusPageUrl(metadata.getStatusPageUrl());
            instance.setHealthCheckUrl(metadata.getHealthCheckUrl());
            if(instance.isSecurePortEnabled()) { 
                instance.setSecureHealthCheckUrl(metadata.getSecureHealthCheckUrl());
            }
            Map<String, String> metadataMap = instance.getMetadataMap();
            if (metadataMap.get("management.port") == null) { 
                metadataMap.put("management.port", String.valueOf(metadata.getManagementPort()));
            }
        }
    
        setupJmxPort(instance, jmxPort);
        return instance;
    }

**RefreshableEurekaClientConfiguration.eurekaClient**  
![在这里插入图片描述][7db4e1797b6ed87d607641715345c517.png_pic_center]  
DiscoveryClient是eureka的核心类，而CloudEurekaClient就是DiscoveryClient的子类  
![在这里插入图片描述][8eb800481819350d34dee82f1c0d1913.png_pic_center]  
![在这里插入图片描述][9736a63fecc1842c1fcc3d2153a29923.png_pic_center]  
进入到DiscoveryClient的构造方法，这里面会初始化一些任务

@Inject
    DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
                    Provider<BackupRegistry> backupRegistryProvider) { 
        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;
        // 该eureka实例
        InstanceInfo myInfo = applicationInfoManager.getInfo();
    
        clientConfig = config;
        staticClientConfig = clientConfig;
        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.urlRandomizer = new EndpointUtils.InstanceInfoBasedUrlRandomizer(instanceInfo);
        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()) { 
            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());
    
        // 不注册到eureka server && 不从eureka server获取注册信息服务列表，则就返回了
        if (!config.shouldRegisterWithEureka() && !config.shouldFetchRegistry()) { 
            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());
    
            // 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();
            logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
                        initTimestampMs, this.getApplications().size());
    
            return;  // no need to setup up an network tasks and we are done
        }
    
        try { 
            // default size of 2 - 1 each for heartbeat and cacheRefresh
            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
    
            eurekaTransport = new EurekaTransport();
            scheduleServerEndpointTask(eurekaTransport, args);
    
            AzToRegionMapper azToRegionMapper;
            if (clientConfig.shouldUseDnsForFetchingServiceUrls()) { 
                azToRegionMapper = new DNSBasedAzToRegionMapper(clientConfig);
            } else { 
                azToRegionMapper = new PropertyBasedAzToRegionMapper(clientConfig);
            }
            if (null != remoteRegionsToFetch.get()) { 
                azToRegionMapper.setRegionsToFetch(remoteRegionsToFetch.get().split(","));
            }
            instanceRegionChecker = new InstanceRegionChecker(azToRegionMapper, clientConfig.getRegion());
        } catch (Throwable e) { 
            throw new RuntimeException("Failed to initialize DiscoveryClient!", e);
        }
    
        // 第一次拉取，如果拉取失败就从备份注册中心拉取
        if (clientConfig.shouldFetchRegistry() && !fetchRegistry(false)) { 
            fetchRegistryFromBackup();
        }
    
        // call and execute the pre registration handler before all background tasks (inc registration) is started
        if (this.preRegistrationHandler != null) { 
            this.preRegistrationHandler.beforeRegistration();
        }
    
        // 需要注册到eureka server && 在初始化时强制注册（默认false，所以这里的register方法默认不执行）
        if (clientConfig.shouldRegisterWithEureka() && clientConfig.shouldEnforceRegistrationAtInit()) { 
            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);
            }
        }
    
        // 初始化定时任务，服务心跳、服务列表获取等功能在此处完成
        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();
        logger.info("Discovery Client initialized at timestamp {} with initial instances count: {}",
                    initTimestampMs, this.getApplications().size());
    }

// 初始化eureka client所有的定时任务
    private void initScheduledTasks() { 
        if (clientConfig.shouldFetchRegistry()) { 
            // 缓存刷新定时任务，拉取注册信息
            int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
            int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
            scheduler.schedule(
                new TimedSupervisorTask(
                    "cacheRefresh",
                    scheduler,
                    cacheRefreshExecutor,
                    registryFetchIntervalSeconds,
                    TimeUnit.SECONDS,
                    expBackOffBound,
                    new CacheRefreshThread()
                ),
                registryFetchIntervalSeconds, TimeUnit.SECONDS);
        }
    
        if (clientConfig.shouldRegisterWithEureka()) { 
            int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
            int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
            logger.info("Starting heartbeat executor: " + "renew interval is: {}", renewalIntervalInSecs);
    
            // 发送心跳定时器
            scheduler.schedule(
                new TimedSupervisorTask(
                    "heartbeat",
                    scheduler,
                    heartbeatExecutor,
                    renewalIntervalInSecs,
                    TimeUnit.SECONDS,
                    expBackOffBound,
                    new HeartbeatThread()
                ),
                renewalIntervalInSecs, TimeUnit.SECONDS);
    
            // 实例信息复制器，不知道为啥叫复制器，里面其实是注册的逻辑
            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()) { 
                        // 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");
        }
    }

[7db4e1797b6ed87d607641715345c517.png_pic_center]: /images/20220828/20aec189909643de838f051a780de65e.png
[8eb800481819350d34dee82f1c0d1913.png_pic_center]: /images/20220828/b0a1687de4d5477cb6149a4f73c169ce.png
[9736a63fecc1842c1fcc3d2153a29923.png_pic_center]: /images/20220828/d395762ade524d2088184052f60df9d1.png