【一起學原始碼-微服務】Nexflix Eureka 原始碼三：EurekaServer啟動之EurekaServer上下文EurekaClient建立

前言

上篇文章已經介紹了 Eureka Server 環境和上下文初始化的一些程式碼，其中重點講解了environment初始化使用的單例模式，以及EurekaServerConfigure基於介面對外暴露配置方法的設計方式。這一講就是講解Eureka Server上下文初始化剩下的內容：Eureka Client初始化。

如若轉載請標明來源：一枝花算不算浪漫

EurekaServer上下文構建之Client

EurekaClientConfigure建立過程

因為eurekaSever是叢集部署的，所以每個eurekaServer都需要註冊到其他註冊中心節點。這裡自己既是一個eurekaServer，也是一個eurekaClient。

擷取EurekaServer中初始化上下文程式碼：

// 3、初始化eureka-server內部的一個eureka-client（用來跟其他的eureka-server節點做註冊和通訊）
// 類的開頭已經說明了：EurekaInstanceConfig其實就是eureka client相關的配置類
if (eurekaClient == null) {
    EurekaInstanceConfig instanceConfig = isCloud(ConfigurationManager.getDeploymentContext())
            ? new CloudInstanceConfig()
            : new MyDataCenterInstanceConfig();
    
    applicationInfoManager = new ApplicationInfoManager(
            instanceConfig, new EurekaConfigBasedInstanceInfoProvider(instanceConfig).get());

    // DefaultEurekaClientConfig類似於上面的DefaultEurekaServerConfig類實現
    EurekaClientConfig eurekaClientConfig = new DefaultEurekaClientConfig();
    eurekaClient = new DiscoveryClient(applicationInfoManager, eurekaClientConfig);
} else {
    applicationInfoManager = eurekaClient.getApplicationInfoManager();
}

再看下eurekaClientConfig建立的程式碼：

public DefaultEurekaClientConfig(String namespace) {
    this.namespace = namespace.endsWith(".")
            ? namespace
            : namespace + ".";

    this.configInstance = Archaius1Utils.initConfig(CommonConstants.CONFIG_FILE_NAME);
    this.transportConfig = new DefaultEurekaTransportConfig(namespace, configInstance);
}

public static DynamicPropertyFactory initConfig(String configName) {

    DynamicPropertyFactory configInstance = DynamicPropertyFactory.getInstance();
    /**
     * 獲取eureka client配置檔案，類似於 {@link DefaultEurekaServerConfig}中的：
     * String eurekaPropsFile = EUREKA_PROPS_FILE.get();
     * private static final DynamicStringProperty EUREKA_PROPS_FILE = DynamicPropertyFactory
     *             .getInstance().getStringProperty("eureka.server.props","eureka-server");
     */
    DynamicStringProperty EUREKA_PROPS_FILE = configInstance.getStringProperty("eureka.client.props", configName);

    String env = ConfigurationManager.getConfigInstance().getString(EUREKA_ENVIRONMENT, "test");
    ConfigurationManager.getConfigInstance().setProperty(ARCHAIUS_DEPLOYMENT_ENVIRONMENT, env);

    String eurekaPropsFile = EUREKA_PROPS_FILE.get();
    try {
        ConfigurationManager.loadCascadedPropertiesFromResources(eurekaPropsFile);
    } catch (IOException e) {
        logger.warn(
                "Cannot find the properties specified : {}. This may be okay if there are other environment "
                        + "specific properties or the configuration is installed with a different mechanism.",
                eurekaPropsFile);

    }

    return configInstance;
}

看到上面程式碼想到了什麼？這完全跟EurekaServerConfig建立的邏輯一樣的呀，程式碼和DefaultEurekaServerConfig一致的邏輯。最後都是交給ConfigurationManager來管理。

EurekaClient建立過程

接著再來看eurekaClient = new DiscoveryClient(applicationInfoManager, eurekaClientConfig);程式碼：

這段程式碼確實很長，我們一段段來解讀，解讀完後再看程式碼：

基於ApplicationInfoManager（包含了服務例項的資訊、配置，作為服務例項管理的一個元件），eureka client相關的配置，一起構建了一個EurekaClient。
這裡有兩個配置：config.shouldFetchRegistry()和config.shouldRegisterWithEureka()

config.shouldFetchRegistry()：
是否需要註冊到別的註冊中心。eurekaServer有個配置：eureka.client.fetchRegistry，單機情況下為false。false表示自己就是註冊中心。我的職責就是維護服務例項，並不需要去檢索服務

config.shouldRegisterWithEureka()：
是否要向別的註冊中心註冊自己。eurekaServer有個配置：eureka.client.registerWithEureka，單機情況下為false。false表示自己不需要向註冊中心註冊自己
建立執行緒池排程任務
建立一個心跳執行緒池
建立一個快取重新整理執行緒池
初始化執行緒排程任務

具體程式碼如下，添加了一些程式碼備註：

@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;
    InstanceInfo myInfo = applicationInfoManager.getInfo();

    clientConfig = config;
    staticClientConfig = clientConfig;
    transportConfig = config.getTransportConfig();
    instanceInfo = myInfo;
    if (myInfo != null) {
        // AppName是服務名稱，instanceInfo.getId就是服務例項id，類似於：ServiceA/0001
        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(","));

    // 是否需要註冊到別的註冊中心。eurekaServer有個配置：eureka.client.fetchRegistry，單機情況下為false。false表示自己就是註冊中心。我的職責就是維護服務例項，並不需要去檢索服務
    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;
    }
    // eurekaServer有個配置：eureka.client.registerWithEureka，單機情況下為false。false表示自己不需要向註冊中心註冊自己
    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());

    // 不需要註冊也不需要抓取 釋放不必要的資源
    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());

        // 建立一個心跳檢查的執行緒池，最大執行緒數為5
        heartbeatExecutor = new ThreadPoolExecutor(
                1, clientConfig.getHeartbeatExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>(),
                new ThreadFactoryBuilder()
                        .setNameFormat("DiscoveryClient-HeartbeatExecutor-%d")
                        .setDaemon(true)
                        .build()
        );  // use direct handoff

        // 支援快取重新整理的執行緒池，最大執行緒數為5
        cacheRefreshExecutor = new ThreadPoolExecutor(
                1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
                new SynchronousQueue<Runnable>(),
                new ThreadFactoryBuilder()
                        .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
                        .setDaemon(true)
                        .build()
        );  // use direct handoff

        // 支援底層的eureka client跟eureka server進行網路通訊的元件
        eurekaTransport = new EurekaTransport();
        // 傳送http請求，呼叫restful介面
        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();
    }
    // 初始化排程任務
    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());
}

/**
 * Initializes all scheduled tasks.
 */
private void initScheduledTasks() {
    // 抓取登錄檔的定時任務，
    if (clientConfig.shouldFetchRegistry()) {
        // registry cache refresh timer
        // registryFetchIntervalSeconds預設為30s
        int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
        int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
        // 執行cacheRefreshExecutor排程任務，預設是30s
        scheduler.schedule(
                new TimedSupervisorTask(
                        "cacheRefresh",
                        scheduler,
                        cacheRefreshExecutor,
                        registryFetchIntervalSeconds,
                        TimeUnit.SECONDS,
                        expBackOffBound,
                        new CacheRefreshThread()
                ),
                registryFetchIntervalSeconds, TimeUnit.SECONDS);
    }

    // 如果要將自己註冊到其他註冊中心
    if (clientConfig.shouldRegisterWithEureka()) {
        //  預設也是30s
        int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
        int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
        logger.info("Starting heartbeat executor: " + "renew interval is: " + renewalIntervalInSecs);

        // Heartbeat timer
        // 執行heartbeatExecutor心跳檢查，預設是30s
        scheduler.schedule(
                new TimedSupervisorTask(
                        "heartbeat",
                        scheduler,
                        heartbeatExecutor,
                        renewalIntervalInSecs,
                        TimeUnit.SECONDS,
                        expBackOffBound,
                        new HeartbeatThread()
                ),
                renewalIntervalInSecs, TimeUnit.SECONDS);

        // InstanceInfo replicator
        // 建立服務副本傳播器
        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");
    }
}

總結

如果是eureka server的話，我們在玩兒spring cloud的時候，會將這個fetchRegistry給手動設定為false，如果是eureka server叢集的話，就還是要保持為true。registerWithEureka也要設定為true。

（1）讀取EurekaClientConfig，包括TransportConfig
（2）儲存EurekaInstanceConfig和InstanceInfo
（3）處理了是否要註冊以及抓取登錄檔，如果不要的話，釋放一些資源
（4）支援排程的執行緒池
（5）支援心跳的執行緒池
（6）支援快取重新整理的執行緒池
（7）EurekaTransport，支援底層的eureka client跟eureka server進行網路通訊的元件，對網路通訊元件進行了一些初始化的操作
（8）如果要抓取登錄檔的話，在這裡就會去抓取登錄檔了，但是如果說你配置了不抓取，那麼這裡就不抓取了
（9）初始化排程任務：如果要抓取登錄檔的話，就會註冊一個定時任務，按照你設定的那個抓取的間隔，每隔一定時間（預設是30s），去執行一個CacheRefreshThread，給放那個排程執行緒池裡去了；如果要向eureka server進行註冊的話，會搞一個定時任務，每隔一定時間傳送心跳，執行一個HeartbeatThread；建立了服務例項副本傳播器，將自己作為一個定時任務進行排程；建立了服務例項的狀態變更的監聽器，如果你配置了監聽，那麼就會註冊監聽器

申明

本文章首發自本人部落格：https://www.cnblogs.com/wang-meng 和公眾號：壹枝花算不算浪漫，如若轉載請標明來源!

感興趣的小夥伴可關注個人公眾號：壹枝花算不算浪漫

【一起學原始碼-微服務】Nexflix Eureka 原始碼三：EurekaServer啟動之EurekaServer上下文EurekaClient建立

前言

EurekaServer上下文構建之Client

EurekaClientConfigure建立過程

EurekaClient建立過程

總結

申明

【一起學原始碼-微服務】Nexflix Eureka 原始碼二：EurekaServer啟動之配置檔案載入以及面向介面的配置項讀取

【一起學原始碼-微服務】Nexflix Eureka 原始碼三：EurekaServer啟動之EurekaServer上下文EurekaClient建立

【一起學原始碼-微服務】Nexflix Eureka 原始碼六：在眼花繚亂的程式碼中，EurekaClient是如何註冊的？

【一起學原始碼-微服務】Nexflix Eureka 原始碼七：通過單元測試來Debug Eureka註冊過程

【一起學原始碼-微服務】Nexflix Eureka 原始碼八：EurekaClient登錄檔抓取精妙設計分析！

【一起學原始碼-微服務】Nexflix Eureka 原始碼九：服務續約原始碼分析

【一起學原始碼-微服務】Nexflix Eureka 原始碼十：服務下線及例項摘除，一個client下線到底多久才會被其他例項感知？

【一起學原始碼-微服務】Nexflix Eureka 原始碼十一：EurekaServer自我保護機制竟然有這麼多Bug？

【一起學原始碼-微服務】Nexflix Eureka 原始碼十二：EurekaServer叢集模式原始碼分析

【一起學原始碼-微服務】Nexflix Eureka 原始碼十三：Eureka原始碼解讀完結撒花篇~！

【一起學原始碼-微服務】Ribbon 原始碼一：Ribbon概念理解及Demo除錯

【一起學原始碼-微服務】Ribbon 原始碼二：通過Debug找出Ribbon初始化流程及ILoadBalancer原理分析

【一起學原始碼-微服務】Ribbon 原始碼三：Ribbon與Eureka整合原理分析

【一起學原始碼-微服務】Ribbon 原始碼四：進一步探究Ribbon的IRule和IPing

【一起學原始碼-微服務】Ribbon原始碼五：Ribbon原始碼解讀彙總篇~

【一起學原始碼-微服務】Feign 原始碼一：原始碼初探，通過Demo Debug Feign原始碼

【一起學原始碼-微服務】Feign 原始碼二：Feign動態代理構造過程

【一起學原始碼-微服務】Feign 原始碼三：Feign結合Ribbon實現負載均衡的原理分析

【一起學原始碼-微服務】Hystrix 原始碼一：Hystrix基礎原理與Demo搭建

【一起學原始碼-微服務】Hystrix 原始碼二：Hystrix核心流程：Hystix非降級邏輯流程梳理

【一起學原始碼-微服務】Nexflix Eureka 原始碼三：EurekaServer啟動之EurekaServer上下文EurekaClient建立

前言

EurekaServer上下文構建之Client

EurekaClientConfigure建立過程

EurekaClient建立過程

總結

申明

相關推薦