EurekaClient source code analysis

Background: I was recently researching springCloud and was very curious about the service registry, so I took a look at the source code, and I need to know about it in future interviews, so make a note of it

Note: EurekaClient has a lot of content, so I’ll just analyze the trunk

Prerequisite: The SpringBoot version is 2.1.5.RELEASE and the spring-cloud version is Greenwich.SR1.

Eureka architecture diagram

From this we can see several important concepts:

Eureka can Register, Renew, Cancel and Get Registy to EurekaServer. Below, we will discuss these concepts respectively.

According to the Automatic configuration feature of SpringBoot, We find the org. Springframework. Cloud. Spring – the cloud – netflix – eureka – client. 2.1.1. The spring – the cloud – netflix – eureka – the client – 2.1.1. Meta-inf /spring.factories under release.jar.

The most important thing is to EurekaClientAutoConfiguration the configuration class

Open the configuration class EurekaClientAutoConfiguration capture important configuration:

@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; }
Copy the code

This bean is the configuration information of Eureka. The prefix of ymal configuration is started with eureka. Client. This configuration class has a large number of default eureka Client configurations.

Such as:

1. Default Eureka configuration:

/** * Default Eureka URL. */public static final String DEFAULT_URL = "http://localhost:8761" + DEFAULT_PREFIX + "/";Copy the code

That is, if the URL of the Eureka server is not configured, it is registered to the local 8761 address by default.

2. By default, the configuration interval is pulled from the Eureka server

/** * Indicates how often(in seconds) to fetch the registry information from the eureka * server. */private int registryFetchIntervalSeconds = 30;Copy the code

In other words, the service configuration information is pulled from the Eureka server every 30 seconds by default.

3. Time interval for replicating instance change information to the Eureka server

/** * Indicates how often(in seconds) to replicate instance changes to be replicated to * the eureka server. */private int instanceInfoReplicationIntervalSeconds = 30;Copy the code

There are many configurations you can learn about yourself.

@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.servlet.context-path"."/");   int serverPort = Integer         .valueOf(env.getProperty("server.port", env.getProperty("port"."8080")));   Integer managementPort = env.getProperty("management.server.port", Integer.class); // nullable. String managementContextPath = env .getProperty("management.server.servlet.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
      
        metadataMap = instance.getMetadataMap(); metadataMap.computeIfAbsent("management.port", k -> String.valueOf(metadata.getManagementPort())); } else { // without the metadata the status and health check URLs will not be set // and the status page and health check url paths will not include the // context path so set them here if (StringUtils.hasText(managementContextPath)) { instance.setHealthCheckUrlPath( managementContextPath + instance.getHealthCheckUrlPath()); instance.setStatusPageUrlPath( managementContextPath + instance.getStatusPageUrlPath()); } } setupJmxPort(instance, jmxPort); return instance; }
      ,>Copy the code

This is mainly the eureka client itself some configuration information.

Next up are the two most important beans: DiscoveryClient and EurekaServiceRegistry

@Beanpublic DiscoveryClient discoveryClient(EurekaClient client,      EurekaClientConfig clientConfig) {   returnnew EurekaDiscoveryClient(client, clientConfig); }@Beanpublic EurekaServiceRegistryeurekaServiceRegistry() {   returnnew EurekaServiceRegistry(); }Copy the code

EurekaServiceRegistry is used for service registration and DiscoveryClient is used for service discovery

Look at the EurekaServiceRegistry

We see that there are several methods in this class:

Finally take a look at DiscoveryClient and see that there are two DiscoveryClients, one is a class and one is an interface:

Classes are provided by Netflix and interfaces are provided by springcloud. Back to EurekaClientAutoConfiguration, it in the statement with EurekaDiscoveryClient DiscoveryClient the bean

@Beanpublic DiscoveryClient discoveryClient(EurekaClient client, EurekaClientConfig clientConfig) {   return newEurekaDiscoveryClient(client, clientConfig); }Copy the code

Find EurekaDiscoveryClient

The EurekaDiscoveryClient contains the EurekaClient interface in Netflix, and the default implementation of this interface is DiscoveryClient

In other words, EurekaDiscoveryClient and DiscoveryClient are combined, and the last class to be called is The Netflix class DiscoveryClient. This class is the key to service registry discovery, so we’ll look at this class.

DiscoveryClient

First look at the source code explanation:

This class is used to interact with the Eureka Server. Its main responsibilities are:

C) <em>Cancellation</em> of the lease from <tt>Eureka Server</tt> during shutdown

The important method in the DiscoveryClient constructor is initScheduledTasks();

Fetch Registers: obtains the registration list

if(clientConfig.shouldFetchRegistry()){    //registry cache refresh timer int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds(); int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound(); scheduler.schedule(new TimedSupervisorTask("cacheRefresh" , scheduler, cacheRefreshExecutor, registryFetchIntervalSeconds, TimeUnit.SECONDS, expBackOffBound, new CacheRefreshThread()), registryFetchIntervalSeconds, TimeUnit.SECONDS); }Copy the code

The eureka client starts a scheduler for a certain amount of time (default is 30 seconds). Can pass registryFetchIntervalSeconds configuration) from eureka server pull eureka client configuration information. Take a closer look at the Runnable CacheRefreshThread.

class CacheRefreshThread implements Runnable { 
public void run(a) { 
  refreshRegistry();// Execute this method mainly on time}}Copy the code

Further analysis of refreshRegistry

void refreshRegistry() { try { ... Boolean Success = fetchRegistry(remoteRegionsModified);if (success) {            registrySize = localRegionApps.get().size(); lastSuccessfulRegistryFetchTimestamp = System.currentTimeMillis(); }... } catch (Throwable e) { logger.error("Cannot fetch registry from server", e);    }}Copy the code

FetchRegistry (Boolean forceFullRegistryFetch) method, the main one is getAndStoreFullRegistry(); To send HTTP requests to the registry to get registration information and cache it locally

private void getAndStoreFullRegistry(a) throws Throwable {... Applications apps =null;    EurekaHttpResponse<Applications> httpResponse = clientConfig.getRegistryRefreshSingleVipAddress() == null ? eurekaTransport.queryClient.getApplications(remoteRegionsRef.get()) : eurekaTransport.queryClient.getVip(clientConfig.getRegistryRefreshSingleVipAddress(), remoteRegionsRef.get());    if (httpResponse.getStatusCode() == Status.OK.getStatusCode()) {        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)) {             Localregionapps. set(this.filterandshuffle (apps)); // Cache the pulled information in localRegionApps. logger.debug("Got full registry with apps hashcode {}", apps.getAppsHashCode()); } else { logger.warn("Not updating applications as another thread is updating it already"); }}Copy the code

And that localRegionApps is used to cache the pull information

private final AtomicReference<Applications> localRegionApps = new AtomicReference<Applications>();Copy the code

Renew Renew of the service

Back to the initScheduledTasks() method,

if(clientConfig.shouldRegisterWithEureka()){    ...    // Heartbeat timer    scheduler.schedule(new TimedSupervisorTask("heartbeat", scheduler, heartbeatExecutor, renewalIntervalInSecs, TimeUnit.SECONDS, expBackOffBound, new HeartbeatThread()), renewalIntervalInSecs, TimeUnit.SECONDS); . }else{    logger.info("Not registering with Eureka server per configuration"); }Copy the code

Take a closer look at the Runnable HeartbeatThread

/** * The heartbeat task that renews the lease in the given intervals. */private class HeartbeatThread implements Runnable {    public void run() {        if(renew()) { lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis(); }}}Copy the code

/** * Renew with the eureka service by making the appropriate REST call */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()) {            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;        }        return httpResponse.getStatusCode() == Status.OK.getStatusCode();    } catch (Throwable e) {        logger.error(PREFIX + "{} - was unable to send heartbeat!", appPathIdentifier, e);        return false;    }}Copy the code

The above code sends a heartbeat at regular intervals (30 seconds by default), and if it returns a 404, registers the client with the Eureka Server.

Register service

/** * Register with the eureka service by making the appropriate REST call. */boolean register(a) 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());    }    returnhttpResponse.getStatusCode() == Status.NO_CONTENT.getStatusCode(); }Copy the code

This method is actually in the DiscoveryClient class.

In the DiscoveryClient. InitScheduledTasks (), to have such a code

// InstanceInfo replicatorinstanceInfoReplicator = new InstanceInfoReplicator( this, instanceInfo, clientConfig.getInstanceInfoReplicationIntervalSeconds(), 2); // burstSize... instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());Copy the code

A scheduler is created in the InstanceInfoReplicator constructor. Registration method is invoked by InstanceInfoReplicator. InstanceInfoReplicator. Start (clientConfig getInitialInstanceInfoReplicationIntervalSeco NDS ()) sends the registration request to the registry

public void start(int initialDelayMs) {    if (started.compareAndSet(false.true)) {        instanceInfo.setIsDirty();  // for initial register Future next = scheduler.schedule(this, initialDelayMs, TimeUnit.SECONDS); scheduledPeriodicRef.set(next); }}Copy the code

public void run() {    try {        discoveryClient.refreshInstanceInfo();        Long dirtyTimestamp = instanceInfo.isDirtyWithTime();        if(dirtyTimestamp ! = null) {// Register 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);    }}Copy the code

Cancel: The service is offline:

This method is actually in the DiscoveryClient class.

/** * unregister w/ the eureka service. */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); }}}Copy the code

This method is called through the shutdown() method

/** * Shuts down Eureka Client. Also sends a deregistration request to the * eureka server. */@PreDestroy@Overridepublic synchronized void shutdown(a) {    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(); logger.info("Completed shut down of DiscoveryClient"); }}Copy the code

This method has an annotation @predestroy on it, indicating that it was triggered before the object was destroyed. You can see that the scheduled task is turned off in this method, notifying the Eureka server to go offline.

So that’s my understanding of Eureka Client,

Conclusion:

1. Springcloud integrates Netflix. Through the combination mode, EurekaDiscoveryClient includes EurekaClient, and all core functions are completed by DiscoveryClient.

2. The Eureka client mainly provides the Register, Renew, Cancel, and Get Registy functions.

3. The interaction between the Eureka client and the Server is triggered by a large number of scheduled tasks. The interaction is triggered by the HTTP protocol.

4. When DiscoveryClient is created, in its constructor, three thread pools are started, and then three timer tasks are started: register the current service to the registry; Continuously send heartbeat for renewal task; Periodically refresh the registry details locally, so that these tasks are executed when the project starts, so to speak.