解读Eureka的TimedSupervisorTask类(自动调节间隔的周期性任务)
作者:程序员欣宸
在Eureka客户端中,尽管ScheduledExecutorService的schedule方法创建的是一次性任务,但通过在任务执行完毕后再次调用schedule方法实现了周期性执行,这种设计既考虑到了任务超时导致的间隔时间调整,又通过CAS实现了多线程同步,展现了简洁而巧妙的设计思想
起因
一个基于Spring Cloud框架的应用,如果注册到了Eureka server,那么它就会定时更新服务列表,
这个定时任务启动的代码在com.netflix.discovery.DiscoveryClient类的initScheduledTasks方法中,
如下(来自工程eureka-client,版本1.7.0):
private void initScheduledTasks() { //更新服务列表 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); } ... //略去其他代码
上述代码中,scheduler是ScheduledExecutorService接口的实现,其schedule方法的官方文档如下所示:
上图红框显示:
该方法创建的是一次性任务,但是在实际测试中,如果在CacheRefreshThread类的run方法中打个断点,就会发现该方法会被周期性调用;
因此问题就来了:
方法schedule(Callable callable,long delay,TimeUnit unit)创建的明明是个一次性任务,但CacheRefreshThread被周期性执行了;
寻找答案
打开的run方法源码,请注意下面的中文注释:
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是个很有用的变量,后面会用到,这里记得每次执行任务成功都会将delay重置 delay.set(timeoutMillis); threadPoolLevelGauge.set((long) executor.getActiveCount()); } catch (TimeoutException e) { logger.error("task supervisor timed out", e); timeoutCounter.increment(); long currentDelay = delay.get(); //任务线程超时的时候,就把delay变量翻倍,但不会超过外部调用时设定的最大延时时间 long newDelay = Math.min(maxDelay, currentDelay * 2); //设置为最新的值,考虑到多线程,所以用了CAS delay.compareAndSet(currentDelay, newDelay); } catch (RejectedExecutionException e) { //一旦线程池的阻塞队列中放满了待处理任务,触发了拒绝策略,就会将调度器停掉 if (executor.isShutdown() || scheduler.isShutdown()) { logger.warn("task supervisor shutting down, reject the task", e); } else { logger.error("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.error("task supervisor threw an exception", e); } throwableCounter.increment(); } finally { //这里任务要么执行完毕,要么发生异常,都用cancel方法来清理任务; if (future != null) { future.cancel(true); } //只要调度器没有停止,就再指定等待时间之后在执行一次同样的任务 if (!scheduler.isShutdown()) { //这里就是周期性任务的原因:只要没有停止调度器,就再创建一次性任务,执行时间时dealy的值, //假设外部调用时传入的超时时间为30秒(构造方法的入参timeout),最大间隔时间为50秒(构造方法的入参expBackOffBound) //如果最近一次任务没有超时,那么就在30秒后开始新任务, //如果最近一次任务超时了,那么就在50秒后开始新任务(异常处理中有个乘以二的操作,乘以二后的60秒超过了最大间隔50秒) scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS); } } }
真相就在上面的最后一行代码中:
scheduler.schedule(this, delay.get(), TimeUnit.MILLISECONDS)
执行完任务后,会再次调用schedule方法,在指定的时间之后执行一次相同的任务,这个间隔时间和最近一次任务是否超时有关,如果超时了就间隔时间就会变大;
总结
从整体上看,TimedSupervisorTask是固定间隔的周期性任务,一旦遇到超时就会将下一个周期的间隔时间调大,如果连续超时,那么每次间隔时间都会增大一倍,一直到达外部参数设定的上限为止,一旦新任务不再超时,间隔时间又会自动恢复为初始值,另外还有CAS来控制多线程同步,简洁的代码,巧妙的设计,值得我们学习;
以上为个人经验,希望能给大家一个参考,也希望大家多多支持脚本之家。