TimedSupervisorTask类（自动调节间隔的周期性任务）

Eureka中客户端缓存刷新、心跳监测，调远程服务端接口操作的定时任务，调远程网络不确定性导致定时任务执行不确定性，Eureka设计了一个自动调节间隔的周期性任务！！！通过一个线程类的run方法的finally中再次调用该线程实现。

1、该定时任务的启动代码：

DiscoveryClient类里的initScheduledTasks()方法：该方法有两处使用了该定时任务线程类。

    private void initScheduledTasks() {if (clientConfig.shouldFetchRegistry()) {// registry cache refresh timerint 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);// Heartbeat timerscheduler.schedule(new TimedSupervisorTask("heartbeat",scheduler,heartbeatExecutor,renewalIntervalInSecs,TimeUnit.SECONDS,expBackOffBound,new HeartbeatThread()//该线程执行续约的具体逻辑),renewalIntervalInSecs, TimeUnit.SECONDS);…………省略}

上述代码中，scheduler是ScheduledExecutorService接口的实现：因为两处使用，所以创建了2个线程的线程池。且在new子线程的时候将scheduler传入。

    scheduler = Executors.newScheduledThreadPool(2,new ThreadFactoryBuilder().setNameFormat("DiscoveryClient-%d").setDaemon(true).build());

其schedule方法的官方文档如下所示：

创建了一个一次性动作，那是怎么实现循环定时执行的？答案就在TimedSupervisorTask线程类的run方法里》

2、线程类的实现代码：

亮点就在run方法的finally里的

scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS);

package com.netflix.discovery;import java.util.TimerTask;
import java.util.concurrent.Future;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;import com.netflix.servo.monitor.Counter;
import com.netflix.servo.monitor.LongGauge;
import com.netflix.servo.monitor.MonitorConfig;
import com.netflix.servo.monitor.Monitors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/*** A supervisor task that schedules subtasks while enforce a timeout.* Wrapped subtasks must be thread safe.** @author David Qiang Liu*/
public class TimedSupervisorTask extends TimerTask {private static final Logger logger = LoggerFactory.getLogger(TimedSupervisorTask.class);private final Counter timeoutCounter;private final Counter rejectedCounter;private final Counter throwableCounter;private final LongGauge threadPoolLevelGauge;private final ScheduledExecutorService scheduler;private final ThreadPoolExecutor executor;private final long timeoutMillis;private final Runnable task;private final AtomicLong delay;private final long maxDelay;public TimedSupervisorTask(String name, ScheduledExecutorService scheduler, ThreadPoolExecutor executor,int timeout, TimeUnit timeUnit, int expBackOffBound, Runnable task) {this.scheduler = scheduler;this.executor = executor;this.timeoutMillis = timeUnit.toMillis(timeout);this.task = task;this.delay = new AtomicLong(timeoutMillis);this.maxDelay = timeoutMillis * expBackOffBound;// Initialize the counters and register.timeoutCounter = Monitors.newCounter("timeouts");rejectedCounter = Monitors.newCounter("rejectedExecutions");throwableCounter = Monitors.newCounter("throwables");threadPoolLevelGauge = new LongGauge(MonitorConfig.builder("threadPoolUsed").build());Monitors.registerObject(name, this);}@Overridepublic 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是个关键变量，后面会用到，//这里每次执行任务成功都会将delay重置,即时间间隔还原到最初值。delay.set(timeoutMillis);threadPoolLevelGauge.set((long) executor.getActiveCount());} catch (TimeoutException e) {logger.warn("task supervisor timed out", e);timeoutCounter.increment();long currentDelay = delay.get();//任务线程超时的时候，就把delay变量翻倍，但不会超过外部调用时设定的最大延时时间long newDelay = Math.min(maxDelay, currentDelay * 2);//设置为最新的值，考虑到多线程，所以用了CASdelay.compareAndSet(currentDelay, newDelay);} catch (RejectedExecutionException e) {if (executor.isShutdown() || scheduler.isShutdown()) {logger.warn("task supervisor shutting down, reject the task", e);} else {logger.warn("task supervisor rejected the task", e);}rejectedCounter.increment();} catch (Throwable e) {//一旦线程池的阻塞队列中放满了待处理任务，触发了拒绝策略，就会将调度器停掉if (executor.isShutdown() || scheduler.isShutdown()) {logger.warn("task supervisor shutting down, can't accept the task");} else {logger.warn("task supervisor threw an exception", e);}throwableCounter.increment();} finally {//这里任务要么执行完毕，要么发生异常，都用cancel方法来清理任务；if (future != null) {future.cancel(true);}//只要调度器没有停止，就在指定等待时间之后再执行一次同样的任务if (!scheduler.isShutdown()) {
//假设外部调用时传入的超时时间为30秒（构造方法的入参timeout），最大间隔时间为50秒(构造方法的入参expBac kOffBound) 
//如果最近一次任务没有超时，那么就在30秒后开始新任务，
//如果最近一次任务超时了，那么就在50秒后开始新任务（异常处理中有个乘以二的操作，乘以二后的60秒超过了最大 间隔50秒）// TODO 亮点所在！！！再次调用。scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS);}}}
}

TimerTask是Java的util包中Runnable的实现类：

public abstract class TimerTask implements Runnable {

真相就在上面的最后一行代码中：scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS)：执行完任务后，会再次调用schedule方法，在指定的时间之后执行一次相同的任务，这个间隔时间和最近一次任务是否超时有关，如果超时了间隔时间就会变大。

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！