Eureka的TimedSupervisorTask类(自动调节间隔的周期性任务)

举报
程序员欣宸 发表于 2022/06/21 07:29:00 2022/06/21
【摘要】 了解Eureka server在定时更新服务列表时依赖的细节

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码):https://github.com/zq2599/blog_demos

起因

  • 一个基于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方法的官方文档如下所示:

image.png

  • 上图红框显示:该方法创建的是一次性任务,但是在实际测试中,如果在CacheRefreshThread类的run方法中打个断点,就会发现该方法会被周期性调用;

  • 因此问题就来了:方法schedule(Callable<V> 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来控制多线程同步,简洁的代码,巧妙的设计,值得我们学习;

欢迎关注华为云博客:程序员欣宸

学习路上,你不孤单,欣宸原创一路相伴…

【版权声明】本文为华为云社区用户原创内容,转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息, 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

举报
请填写举报理由
0/200