SpringCloud源码学习笔记之Eureka客户端——服务续约

1、服务续约入口

  在《Eureka客户端——初始化》一篇中，我们知道，在DiscoveryClient对象的构造函数中的initScheduledTasks()方法中，实现了服务续约。具体实现如下：

 // 创建心跳服务线程，同时进行服务续约
 heartbeatTask = new TimedSupervisorTask(
          "heartbeat",
          scheduler,
          heartbeatExecutor,
          renewalIntervalInSecs,
          TimeUnit.SECONDS,
          expBackOffBound,
          new HeartbeatThread()
  );
  //执行定时任务
  scheduler.schedule(
          heartbeatTask,
          renewalIntervalInSecs, TimeUnit.SECONDS);

  在上述方法中，首先创建了一个TimedSupervisorTask任务，其中实际执行的任务是由HeartbeatThread实现，然后再调用scheduler.schedule()方法，定时执行续约任务。

2、TimedSupervisorTask类TimerTask

  其实，服务续约和服务发现的逻辑类似，不过在学习服务续约的时候，没有深入了解TimedSupervisorTask类，我们这里先来学习一下该类。TimedSupervisorTask类继承了TimerTask类，所以在调度器执行schedule()方法时，TimedSupervisorTask对象的run()方法就会被执行。该类主要为了监督子任务执行，即HeartbeatThread对象的任务，同时增强了超时处理、线程安全等。

2.1、构造函数

public TimedSupervisorTask(String name, ScheduledExecutorService scheduler, ThreadPoolExecutor executor,int timeout, TimeUnit timeUnit, int expBackOffBound, Runnable task) {
    //线程名称               
    this.name = name;
    //线程调度器
    this.scheduler = scheduler;
    //心跳线程池执行器，ThreadPoolExecutor对象
    this.executor = executor;
    //使用续约间隔，作为超期时间
    this.timeoutMillis = timeUnit.toMillis(timeout);
    //实际续约需要执行的任务，HeartbeatThread对象
    this.task = task;
    //使用续约间隔，作为延期执行时间
    this.delay = new AtomicLong(timeoutMillis);
    //超期重试的最大时间，其中expBackOffBound表示心跳超时重试延迟时间的最大乘数值
    this.maxDelay = timeoutMillis * expBackOffBound;

    // 心跳结果的计数器
    successCounter = Monitors.newCounter("success");
    timeoutCounter = Monitors.newCounter("timeouts");
    rejectedCounter = Monitors.newCounter("rejectedExecutions");
    throwableCounter = Monitors.newCounter("throwables");
    threadPoolLevelGauge = new LongGauge(MonitorConfig.builder("threadPoolUsed").build());
    Monitors.registerObject(name, this);
}

2.2、run()方法

@Override
public void run() {
	//保存异步执行结果的变量
    Future<?> future = null;
    try {
    	//提交任务，并把结果保存到future变量中
        future = executor.submit(task);
        threadPoolLevelGauge.set((long) executor.getActiveCount());
        //设置超时时间
        future.get(timeoutMillis, TimeUnit.MILLISECONDS);  // block until done or timeout
        delay.set(timeoutMillis);
        threadPoolLevelGauge.set((long) executor.getActiveCount());
        //计数
        successCounter.increment();
    } 
    
	//省略各类续约结构的计数和日志 逻辑 ……
	
	 finally {
        if (future != null) {//取消
            future.cancel(true);
        }

        if (!scheduler.isShutdown()) {//执行下一次调用，通过这种方式实现心跳的定期发送
            scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS);
        }
    }
}

3、HeartbeatThread类

  通过前面学习，我们知道了HeartbeatThread类是实现心跳或续约的核心逻辑，该类是定义在DiscoveryClient类中的一个内部类。该类也是Runnable 接口的实现类，所以在执行的时候，会调用run()方法。具体实现如下：

private class HeartbeatThread implements Runnable {

     public void run() {
         if (renew()) {
             lastSuccessfulHeartbeatTimestamp = System.currentTimeMillis();
         }
     }
 }

  通过HeartbeatThread类的定义，我们可以知道，在run()方法中，首先是调用了DiscoveryClient类的renew()方法，进行续约，然后如果续约成功，则会修改lastSuccessfulHeartbeatTimestamp变量的值，即把当前时间作为上次续约成功的时间。

4、DiscoveryClient类的renew()方法

  在该方法中，和服务注册类似，通过调用AbstractJerseyEurekaHttpClient类的sendHeartBeat()方法发送心跳数据，然后根据返回结果再进行下一步处理。如果心跳续约返回NOT_FOUND状态，则调用register()方法重新进行续约。具体实现如下：

boolean renew() {
    EurekaHttpResponse<InstanceInfo> httpResponse;
    try {
    	//通过调用AbstractJerseyEurekaHttpClient类的sendHeartBeat()方法发送心跳数据
        httpResponse = eurekaTransport.registrationClient.sendHeartBeat(instanceInfo.getAppName(), instanceInfo.getId(), instanceInfo, null);
        logger.debug(PREFIX + "{} - Heartbeat status: {}", appPathIdentifier, httpResponse.getStatusCode());
        //如果心跳续约返回NOT_FOUND状态，则调用register()方法重新进行续约
        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;
    }
}

5、AbstractJerseyEurekaHttpClient类的sendHeartBeat()方法

  和AbstractJerseyEurekaHttpClient类的register()方法方法类似，还是发送了一个http请求。

@Override
public EurekaHttpResponse<InstanceInfo> sendHeartBeat(String appName, String id, InstanceInfo info, InstanceStatus overriddenStatus) {
   String urlPath = "apps/" + appName + '/' + id;
   ClientResponse response = null;
   try {
       WebResource webResource = jerseyClient.resource(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();
       addExtraHeaders(requestBuilder);
       response = requestBuilder.put(ClientResponse.class);
       EurekaHttpResponseBuilder<InstanceInfo> eurekaResponseBuilder = anEurekaHttpResponse(response.getStatus(), InstanceInfo.class).headers(headersOf(response));
       if (response.hasEntity() &&
               !HTML.equals(response.getType().getSubtype())) { //don't try and deserialize random html errors from the server
           eurekaResponseBuilder.entity(response.getEntity(InstanceInfo.class));
       }
       return eurekaResponseBuilder.build();
   } finally {
       if (logger.isDebugEnabled()) {
           logger.debug("Jersey HTTP PUT {}/{}; statusCode={}", serviceUrl, urlPath, response == null ? "N/A" : response.getStatus());
       }
       if (response != null) {
           response.close();
       }
   }
}