为什么要单独讲解TimedSupervisorTask这个类呢?因为这个类在我们DiscoveryClient类的initScheduledTasks方法进行定时任务初始化时被使用得比较多,所以我们需要了解下这个类,我们先看下TimedSupervisorTask这个类在initScheduledTasks的具体使用:
private final ScheduledExecutorService scheduler;
private void initScheduledTasks() {
…省略其他代码
// 初始化定时拉取服务注册信息
scheduler.schedule(
new TimedSupervisorTask(
"cacheRefresh",
scheduler,
cacheRefreshExecutor,
registryFetchIntervalSeconds,
TimeUnit.SECONDS,
expBackOffBound,
new CacheRefreshThread()
),
registryFetchIntervalSeconds, TimeUnit.SECONDS);
…省略其他代码
// 初始化定时服务续约任务
scheduler.schedule(
new TimedSupervisorTask(
"heartbeat",
scheduler,
heartbeatExecutor,
renewalIntervalInSecs,
TimeUnit.SECONDS,
expBackOffBound,
new HeartbeatThread()
),
renewalIntervalInSecs, TimeUnit.SECONDS);
…省略其他代码
}
由此可见,TimedSupervisorTask类被使用在了定时任务的初始化中,我们具体来看看这个类的结构:
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);
}
@Override
public void run() {
Future<?> future = null;
try {
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());
} catch (TimeoutException e) {
logger.warn("task supervisor timed out", e);
timeoutCounter.increment();
long currentDelay = delay.get();
// 如果出现异常,则将时间*2,然后取 定时时间 和 最长定时时间中最小的为下次任务执行的延时时间
long newDelay = Math.min(maxDelay, currentDelay * 2);
delay.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 {
if (future != null) {
future.cancel(true);
}
if (!scheduler.isShutdown()) {
scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS);
}
}
}
}
我们可以仔细看看run方法的具体实现,因为这里有一个值得借鉴的设计思路!!!
**我们简单来看看这个方法具体执行流程:
1.执行submit()方法提交任务
2.执行future.get()方法,如果没有在规定的时间得到返回值或者任务出现异常,则进入异常处理catch代码块。
3.如果发生异常
a. 发生TimeoutException异常,则执行Math.min(maxDelay, currentDelay ️ 2);得到任务延时时间 ️ 2 和 最大延时时间的最小值,然后改变任务的延时时间timeoutMillis(延时任务时间默认值是30s)
b.发生RejectedExecutionException异常,则将rejectedCounter值+1
c.发生Throwable异常,则将throwableCounter值+1
4.如果没有发生异常,则再设置一次延时任务时间timeoutMillis
5.进入finally代码块
a.如果future不为null,则执行future.cancel(true),中断线程停止任务
b.如果线程池没有shutdown,则创建一个新的定时任务**
\(\color{red}{注意}\):不知道有没有小伙伴发现,不管我们的定时任务执行是成功还是结束(如果还没有执行结束,也会被中断),然后会再重新初始化一个新的任务。并且这个任务的延时时间还会因为不同的情况受到改变,在try代码块中如果不发现异常,则会重新初始化延时时间,如果发生TimeoutException异常,则会更改延时时间,更改为 任务延时时间 ️ 2 和 最大延时时间的最小值。所以我们会发现这样的设计会让整个延时任务很灵活。如果不发生异常,则延时时间不会变;如果发现异常,则增长延时时间;如果程序又恢复正常了,则延时时间又恢复成了默认值。
总结:我们在设计延时/周期性任务时就可以参考TimedSupervisorTask的实现,程序一旦遇到发生超时异常,就将间隔时间调大,如果连续超时,那么每次间隔时间都会增大一倍,一直到达外部参数设定的上限为止,一旦新任务不再发生超时异常,间隔时间又会自动恢复为初始值。
手机扫一扫
移动阅读更方便
你可能感兴趣的文章