Eureka源码深度解析（2）

不念不忘少年蓝@ 2022-02-04 17:43 346阅读 0赞

### DiscoveryClient 核心方法解析 ###

** 1）服务注册（发送注册请求到注册中心）**

boolean register() throws Throwable {
                ...
                EurekaHttpResponse<Void> httpResponse;
                try {
                    //主要的注册功能 实现在AbstractJerseyEurekaHttpClient.register()
                    httpResponse = eurekaTransport.registrationClient.register(instanceInfo);
                }
                ...
                return httpResponse.getStatusCode() == 204;
            }
         
            //AbstractJerseyEurekaHttpClient.register()
            public EurekaHttpResponse<Void> register(InstanceInfo info) {
                String urlPath = "apps/" + info.getAppName();
                ClientResponse response = null;
                EurekaHttpResponse var5;
    
                try {
                    // 1.构造一个HTTP请求
                    Builder resourceBuilder = jerseyClient.resource(serviceUrl).path(urlPath).getRequestBuilder();
                    addExtraHeaders(resourceBuilder);
                    
                    // 2.封装请求类型和返回类型，将当前服务的元信息封装为InstanceInfo，
                    // 发送post请求到serviceUrl，serviceUrl即我们在配置文件中配置的defaultZone
                    response = resourceBuilder
                            .header("Accept-Encoding", "gzip")
                            .type(MediaType.APPLICATION_JSON_TYPE)
                            .accept(MediaType.APPLICATION_JSON)
                            .post(ClientResponse.class, info);
                    
                    // 3.返回响应状态
                    var5 = EurekaHttpResponse.anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
    
                } finally {
                    if (logger.isDebugEnabled()) {
                        logger.debug("Jersey HTTP POST {}/{} with instance {}; statusCode={}", serviceUrl, urlPath, info.getId(),
                                response == null ? "N/A" : response.getStatus());
                    }
                    if (response != null) {
                        response.close();
                    }
                }
    
                return var5;
            }

**    2）服务续约（本质就是发送当前应用的心跳请求到注册中心）**

boolean renew() {
            try {
                // 1.本质就是发送心跳请求
                // 2.真正实现为 AbstractJerseyEurekaHttpClient.sendHeartBeat()
                EurekaHttpResponse<InstanceInfo> httpResponse = this.eurekaTransport.registrationClient.sendHeartBeat(this.instanceInfo.getAppName(), this.instanceInfo.getId(), this.instanceInfo, (InstanceStatus)null);
                logger.debug("DiscoveryClient_{} - Heartbeat status: {}", this.appPathIdentifier, httpResponse.getStatusCode());
                
                // 2.如果请求失败，则调用注册服务请求
                if (httpResponse.getStatusCode() == 404) {
                    this.REREGISTER_COUNTER.increment();
                    logger.info("DiscoveryClient_{} - Re-registering apps/{}", this.appPathIdentifier, this.instanceInfo.getAppName());
                    long timestamp = this.instanceInfo.setIsDirtyWithTime();
                    boolean success = this.register();
                    if (success) {
                        this.instanceInfo.unsetIsDirty(timestamp);
                    }
    
                    return success;
                } else {
                    return httpResponse.getStatusCode() == 200;
                }
            } catch (Throwable var5) {
                logger.error("DiscoveryClient_{} - was unable to send heartbeat!", this.appPathIdentifier, var5);
                return false;
            }
        }
     
     
        // AbstractJerseyEurekaHttpClient.sendHeartBeat()
        public EurekaHttpResponse<InstanceInfo> sendHeartBeat(String appName, String id, InstanceInfo info, InstanceStatus overriddenStatus) {
            String urlPath = "apps/" + appName + '/' + id;
            ClientResponse response = null;
    
            EurekaHttpResponse var10;
            try {
                // 主要就是将当前实例的元信息（InstanceInfo）以及状态（UP）通过HTTP请求发送到serviceUrl
                WebResource webResource = this.jerseyClient.resource(this.serviceUrl).path(urlPath).queryParam("status", info.getStatus().toString()).queryParam("lastDirtyTimestamp", info.getLastDirtyTimestamp().toString());
                if (overriddenStatus != null) {
                    webResource = webResource.queryParam("overriddenstatus", overriddenStatus.name());
                }
    
                Builder requestBuilder = webResource.getRequestBuilder();
                this.addExtraHeaders(requestBuilder);
                response = (ClientResponse)requestBuilder.put(ClientResponse.class);
                EurekaHttpResponseBuilder<InstanceInfo> eurekaResponseBuilder = EurekaHttpResponse.anEurekaHttpResponse(response.getStatus(), InstanceInfo.class).headers(headersOf(response));
                if (response.hasEntity()) {
                    eurekaResponseBuilder.entity(response.getEntity(InstanceInfo.class));
                }
    
                var10 = eurekaResponseBuilder.build();
            } finally {
                if (logger.isDebugEnabled()) {
                    logger.debug("Jersey HTTP PUT {}/{}; statusCode={}", new Object[]{this.serviceUrl, urlPath, response == null ? "N/A" : response.getStatus()});
                }
    
                if (response != null) {
                    response.close();
                }
    
            }
    
            return var10;
        }

**3）服务调用（本质就是获取调用服务名所对应的服务提供者实例信息，包括IP、port等）**

public List<InstanceInfo> getInstancesByVipAddress(String vipAddress, boolean secure) {
            return this.getInstancesByVipAddress(vipAddress, secure, this.instanceRegionChecker.getLocalRegion());
        }
    
        public List<InstanceInfo> getInstancesByVipAddress(String vipAddress, boolean secure, @Nullable String region) {
            if (vipAddress == null) {
                throw new IllegalArgumentException("Supplied VIP Address cannot be null");
            } else {
                Applications applications;
                // 1.判断服务提供方是否当前region，若是的话直接从localRegionApps中获取
                if (this.instanceRegionChecker.isLocalRegion(region)) {
                    applications = (Applications)this.localRegionApps.get();
                } else {
                    // 2.否则的话从远程region获取
                    applications = (Applications)this.remoteRegionVsApps.get(region);
                    if (null == applications) {
                        logger.debug("No applications are defined for region {}, so returning an empty instance list for vip address {}.", region, vipAddress);
                        return Collections.emptyList();
                    }
                }
                // 3.从applications中获取服务名称对应的实例名称列表
                return !secure ? applications.getInstancesByVirtualHostName(vipAddress) : applications.getInstancesBySecureVirtualHostName(vipAddress);
            }
        }

**4）服务下线（本质就是发送取消注册的HTTP请求到注册中心）**

void unregister() {
            if (this.eurekaTransport != null && this.eurekaTransport.registrationClient != null) {
                try {
                    logger.info("Unregistering ...");
                    EurekaHttpResponse<Void> httpResponse = this.eurekaTransport.registrationClient.cancel(this.instanceInfo.getAppName(), this.instanceInfo.getId());
                    logger.info("DiscoveryClient_{} - deregister  status: {}", this.appPathIdentifier, httpResponse.getStatusCode());
                } catch (Exception var2) {
                    logger.error("DiscoveryClient_{} - de-registration failed{}", new Object[]{this.appPathIdentifier, var2.getMessage(), var2});
                }
            }
        }
    
      // AbstractJerseyEurekaHttpClient.cancel()
      public EurekaHttpResponse<Void> cancel(String appName, String id) {
            String urlPath = "apps/" + appName + '/' + id;
            ClientResponse response = null;
    
            EurekaHttpResponse var6;
            try {
                Builder resourceBuilder = this.jerseyClient.resource(this.serviceUrl).path(urlPath).getRequestBuilder();
                this.addExtraHeaders(resourceBuilder);
                // 本质就是发送delete请求到注册中心
                response = (ClientResponse)resourceBuilder.delete(ClientResponse.class);
                var6 = EurekaHttpResponse.anEurekaHttpResponse(response.getStatus()).headers(headersOf(response)).build();
            } finally {
                if (logger.isDebugEnabled()) {
                    logger.debug("Jersey HTTP DELETE {}/{}; statusCode={}", new Object[]{this.serviceUrl, urlPath, response == null ? "N/A" : response.getStatus()});
                }
    
                if (response != null) {
                    response.close();
                }
    
            }
            return var6;
        }

通过以上分析可知，服务的注册、下线等操作实际上就是通过发送HTTP请求到注册中心来实现的。那么这些操作的执行时机是什么时候呢？是什么时候服务注册操作会被调用？下线操作是如何被触发的？我们还是来看com.netflix.discovery.DiscoveryClient这个类，这个类在应用启动的时候被加载到容器中，肯定会调用其构造方法，

### DiscoveryClient  **构造方法解析** ###

@Inject
        DiscoveryClient(ApplicationInfoManager applicationInfoManager, EurekaClientConfig config, AbstractDiscoveryClientOptionalArgs args,
                        Provider<BackupRegistry> backupRegistryProvider) {
            // 一系列的参数配置
            ...
     
            try {
                // 下面是几个线程池的创建
                scheduler = Executors.newScheduledThreadPool(3,
                        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()
                ); 
                
                cacheRefreshExecutor = new ThreadPoolExecutor(
                        1, clientConfig.getCacheRefreshExecutorThreadPoolSize(), 0, TimeUnit.SECONDS,
                        new SynchronousQueue<Runnable>(),
                        new ThreadFactoryBuilder()
                                .setNameFormat("DiscoveryClient-CacheRefreshExecutor-%d")
                                .setDaemon(true)
                                .build()
                );
                ...
               
            // 核心方法调用  启动定时任务
            initScheduledTasks();
            try {
                Monitors.registerObject(this);
            } catch (Throwable e) {
                logger.warn("Cannot register timers", e);
            }
            ...
        }
            
        //initScheduledTasks()
        private void initScheduledTasks() {
            if (clientConfig.shouldFetchRegistry()) {
                // registry cache refresh timer
                int registryFetchIntervalSeconds = clientConfig.getRegistryFetchIntervalSeconds();
                int expBackOffBound = clientConfig.getCacheRefreshExecutorExponentialBackOffBound();
                // 1.执行刷新任务的定时器
                scheduler.schedule(
                        new TimedSupervisorTask(
                                "cacheRefresh",
                                scheduler,
                                cacheRefreshExecutor,
                                registryFetchIntervalSeconds,
                                TimeUnit.SECONDS,
                                expBackOffBound,
                                //线程任务在这里，主要就是为了执行这个Thread的run()
                                new CacheRefreshThread()
                        ),
                        registryFetchIntervalSeconds, TimeUnit.SECONDS);
            }
     
            // 2.客户端默认注册到eureka，shouldRegisterWithEureka()默认为true
            if (clientConfig.shouldRegisterWithEureka()) {
                int renewalIntervalInSecs = instanceInfo.getLeaseInfo().getRenewalIntervalInSecs();
                int expBackOffBound = clientConfig.getHeartbeatExecutorExponentialBackOffBound();
                logger.info("Starting heartbeat executor: " + "renew interval is: " + renewalIntervalInSecs);
     
                // 3.执行心跳任务的定时器
                scheduler.schedule(
                        new TimedSupervisorTask(
                                "heartbeat",
                                scheduler,
                                heartbeatExecutor,
                                renewalIntervalInSecs,
                                TimeUnit.SECONDS,
                                expBackOffBound,
                                // 主要就是执行该Thread的run()
                                new HeartbeatThread()
                        ),
                        renewalIntervalInSecs, TimeUnit.SECONDS);
     
                // 获取当前注册服务的基本信息
                instanceInfoReplicator = new InstanceInfoReplicator(
                        this,
                        instanceInfo,
                        clientConfig.getInstanceInfoReplicationIntervalSeconds(),
                        2); 
     
                ...
     
                // 4.将当前服务注册到注册中心 
                instanceInfoReplicator.start(clientConfig.getInitialInstanceInfoReplicationIntervalSeconds());
            } else {
                logger.info("Not registering with Eureka server per configuration");
            }
        }

**    1）CacheRefreshThread执行刷新任务的线程**

class CacheRefreshThread implements Runnable {
            public void run() {
                refreshRegistry();
            }
        }
     
    
        @VisibleForTesting
        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);
            }        
        }
    
        private boolean fetchRegistry(boolean forceFullRegistryFetch) {
            Stopwatch tracer = this.FETCH_REGISTRY_TIMER.start();
    
            label122: {
                boolean var4;
                try {
                    Applications applications = this.getApplications();
                    if (!this.clientConfig.shouldDisableDelta() && Strings.isNullOrEmpty(this.clientConfig.getRegistryRefreshSingleVipAddress()) && !forceFullRegistryFetch && applications != null && applications.getRegisteredApplications().size() != 0 && applications.getVersion() != -1L) {
                        this.getAndUpdateDelta(applications);
                    } else {
                        logger.info("Disable delta property : {}", this.clientConfig.shouldDisableDelta());
                        logger.info("Single vip registry refresh property : {}", this.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() == -1L);
                        this.getAndStoreFullRegistry();
                    }
    
                    applications.setAppsHashCode(applications.getReconcileHashCode());
                    this.logTotalInstances();
                    break label122;
                } catch (Throwable var8) {
                    logger.error("DiscoveryClient_{} - was unable to refresh its cache! status = {}", new Object[]{this.appPathIdentifier, var8.getMessage(), var8});
                    var4 = false;
                } finally {
                    if (tracer != null) {
                        tracer.stop();
                    }
    
                }
    
                return var4;
            }
    
            this.onCacheRefreshed();
            this.updateInstanceRemoteStatus();
            return true;
        }

**    2）HeartbeatThread执行心跳续约任务的线程**

private class HeartbeatThread implements Runnable {
            private HeartbeatThread() {
            }
    
            public void run() {
                //最终调用renew()续约
                if (DiscoveryClient.this.renew()) {
                    DiscoveryClient.this.lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
                }
    
            }
        }

**3）this.instanceInfoReplicator.start(this.clientConfig.getInitialInstanceInfoReplicationIntervalSeconds()); 发送注册请求到注册中心**

public void start(int initialDelayMs) {
            if (this.started.compareAndSet(false, true)) {
                this.instanceInfo.setIsDirty();
                // 启动一个定时任务，任务指向this，
                // class InstanceInfoReplicator implements Runnable ，由该类可知，真正执行的是其run()方法
                Future next = this.scheduler.schedule(this, (long)initialDelayMs, TimeUnit.SECONDS);
                this.scheduledPeriodicRef.set(next);
            }
        }
    
    public void run() {
            boolean var6 = false;
    
            ScheduledFuture next;
            label53: {
                try {
                    var6 = true;
                    this.discoveryClient.refreshInstanceInfo();
                    Long dirtyTimestamp = this.instanceInfo.isDirtyWithTime();
                    if (dirtyTimestamp != null) {
                        // 真正的注册实现在这里，参考上篇问章可知，这里也是发送一个HTTP请求到注册中心
                        this.discoveryClient.register();
                        this.instanceInfo.unsetIsDirty(dirtyTimestamp);
                        var6 = false;
                    } else {
                        var6 = false;
                    }
                    break label53;
                } catch (Throwable var7) {
                    logger.warn("There was a problem with the instance info replicator", var7);
                    var6 = false;
                } finally {
                    if (var6) {
                        ScheduledFuture next = this.scheduler.schedule(this, (long)this.replicationIntervalSeconds, TimeUnit.SECONDS);
                        this.scheduledPeriodicRef.set(next);
                    }
                }
    
                next = this.scheduler.schedule(this, (long)this.replicationIntervalSeconds, TimeUnit.SECONDS);
                this.scheduledPeriodicRef.set(next);
                return;
            }
    
            next = this.scheduler.schedule(this, (long)this.replicationIntervalSeconds, TimeUnit.SECONDS);
            this.scheduledPeriodicRef.set(next);
        }

综上分析可知，在项目启动的时候通过@`EnableDiscoveryClient最终实例化DiscoveryClient实例`，在其构造方法中，启动了三个线程池， 并且分别启动了三个定时任务：

a、注册当前服务到注册中心；

b、持续发送心跳进行续约任务；

c、定时刷新注册中心注册细信息到本地