优雅的时间轮算法

最近看了分布式任务调度的xxl-job框架的源码,熟悉了整个调度的流程后,对其中使用的时间轮算法很感兴趣,觉得这个算法很有意思,同样值得学习分享

背景

在xxl-job的框架中,在调度任务的时候,首先会将任务下次需要执行的时间nextTriggerTime放入到jobinfo表中,然后后台有一个线程在每隔个5s左右捞取从此刻到将来5秒内需要执行的指定数量的任务,即nextTriggerTime < nowTime + 5s的预读时间,而在捞出的这些任务中由于是未来5s内需要执行的,所有有着一些任务是未来执行的而不是此刻,那该怎么办呢?毕竟捞都已经捞出来了。当然我们可以为每个任务起一个线程,然后让线程等待中间的间隔时间,然后执行,但是如果这样的任务很多的话,就会创建太多的线程,这是不可取的;其实,我们还应该想到的一个方式是使用jdk自带的DelayQueue延迟队列,但是由于该延迟队列底层是变种的Leader-Follower的模式(不明白的可以去了解下DelayQueue的底层),这样当有很多任务需要同时触发时,我们仍然会需要创建很多的线程才能完成,否则就会造成任务的执行延迟。所以我们就想能不能只使用一个线程来完成这件事情的,就这样,时间轮算法诞生了。

xxl-job中的简易时间轮

在xxl-job中,作者是将时间轮分为了60个槽,索引分别是0,1,2...,59,其实就是一分钟的60s,然后每一个槽对应着一个任务List集合(如下图),通过(nextTriggerTime/1000)%60来找到每个任务对应的执行时间秒,由于这些任务都是在将来5秒内需要执行的,所以,这个时间轮中最多有4个槽有数据,这样所有的待执行的任务都放入了时间轮中。任务的执行,需要另一个线程来每秒捞一次时间轮,而作者避免处理耗时太长,导致有一秒的任务被遗落,所以每次都会向前校验一个,比如说,当前需要捞取索引8的槽,那么就会同时尝试去捞取7,8两个槽的数据,如果7号槽有数据证明前面的任务执行慢了导致7号被跳过,此时捞出来执行。这样一个简易的时间轮就实现了。

时间轮简易模型.png

之所以说这是一个简易的时间轮是因为,xxl-job中考虑的情况很简单,因为它是扫描的未来5秒内将要执行的任务,所以以秒为最小单位就可以了,而且不需要考虑执行时间大于一轮的这种情况,如果我们需要考虑这种情况呢?那就接着往下看吧

Netty中的时间轮

1.几个重要的数据结构

时间轮的思想都是一样的,只不过Netty中考虑的更加全面,同样也是值得我们学习的,以后遇到需要使用的情况,我们可以直接使用Netty封装好的这个工具类。
在Netty中时间轮是通过HashedWheelTimer这个类来进行封装的,我们先看一下这个类的常用构造器:

        public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration,           
 TimeUnit unit, int ticksPerWheel) {
        this(threadFactory, tickDuration, unit, ticksPerWheel, true);
    }

其中有两个需要我们自定义的属性值,一个是tickDuration,这个属性是表示时钟多久走动一次,也就是时间间隔,好比xxl-job中时间轮的一秒走一次;两一个是ticksPerWheel,这个属性使用了指定时间轮一起有几个刻度,好比如xxl-job中时间轮的60个刻度。
在HashedWheelTimer中还有两个重要的内部类:HashedWheelBucketHashedWheelTimeout;在说这两个之前,提一下netty中时间轮的数据结构:数组+双向链表;而HashedWheelBucket就是数组中的元素结构,HashedWheelTimeout就是双向链表中的每个任务封装后的结构。其中HashedWheelBucket的结构比较简单,就是头尾指针属性(代码如下)以及一些链表的增删改查操作,

    private static final class HashedWheelBucket {
        // Used for the linked-list datastructure
        private HashedWheelTimeout head;
        private HashedWheelTimeout tail;
    }

而在HashedWheelTimeout的任务封装的结构中,有一个remainingRounds属性,这个属性表示还需要时间轮走几圈后到这个索引时才执行这个任务,也就是说只有任务的remainingRounds属性值为0,当时间轮到达这个任务刻度是才会执行这个任务,否则只会使任务的remainingRounds属性值减一;通过remainingRounds就实现了我们前面所有的当任务执行时间大于一轮的情况。
接下来看看netty中时间轮的工作原理吧:

2.工作原理

当我们创建时间轮对象HashedWheelTimer时,只是创建好了时间轮的数组,工作线程并没有启动,在我们往轮子中添加任务的时候,即调用newTimeout()方法的时候,才会启动工作线程,而且并不是直接将我们的任务丢到时间轮对应的bucket中的,而是先将任务放到一个名为timeouts的Mpsc队列中。我们看下代码:

   @Override
    public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
        ObjectUtil.checkNotNull(task, "task");
        ObjectUtil.checkNotNull(unit, "unit");
        // 等待处理的任务数加一
        long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();

        if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
            pendingTimeouts.decrementAndGet();
            throw new RejectedExecutionException("Number of pending timeouts ("
                + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending "
                + "timeouts (" + maxPendingTimeouts + ")");
        }
        // 首次的话会在此处启动Worker线程,并阻塞等待Worker线程生成好初始时间startTime
        start();

        // Add the timeout to the timeout queue which will be processed on the next tick.
        // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.
        long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;

        // Guard against overflow.
        if (delay > 0 && deadline < 0) {
            deadline = Long.MAX_VALUE;
        }
        // 封装任务
        HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
        // 添加到队列中
        timeouts.add(timeout);
        return timeout;
    }

添加到队列中之后,何时将队列中的任务添加到bucket中的呢?以及任务是何时被执行的呢?要回答这些问题,我们就得来看Worker线程是怎么工作逻辑了,我们来看下其中重要的do-while代码片段:

    do {
                // 控制时钟拨动
                final long deadline = waitForNextTick();
                if (deadline > 0) {
                    // 拿到当前时刻应该操作的bucket索引
                    int idx = (int) (tick & mask);
                    // 处理取消的任务
                    processCancelledTasks();
                    HashedWheelBucket bucket =
                            wheel[idx];
                    // 将队列中的任务放到对应的bucket中
                    transferTimeoutsToBuckets();
                    // 执行该时刻需要执行的任务
                    bucket.expireTimeouts(deadline);
                    // tick值加一,可以与运算获取下一个bucket索引
                    tick++;
                }
            } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);

在waitForNextTick方法中,会判断距离下次时钟拨动还有多久,并会阻塞到时钟拨动的时刻,也就是在这个地方控制了每隔多少时间时钟拨动。
当时钟拨动后,通过起始值为0的tick和时钟刻度数减一的mask(时钟刻度数控制的是2的次方)进行与运算来一直循环遍历整个时钟,此处选择的与运算的效率是比xxl-job中取模运算的性能高的。
在processCancelledTasks方法中会移除那些取消了的任务。
在 transferTimeoutsToBuckets方法中,会将之前放到队列中的任务分配到各个对应的bucket中,看下这块的代码:

            private void transferTimeoutsToBuckets() {
            // 一次从队列中取100000个来进行入bucket处理
            for (int i = 0; i < 100000; i++) {
                HashedWheelTimeout timeout = timeouts.poll();
                if (timeout == null) {
                    // all processed
                    break;
                }
                if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
                    // Was cancelled in the meantime.
                    continue;
                }
                // 计算该任务应该在第几轮处理
                long calculated = timeout.deadline / tickDuration;
                timeout.remainingRounds = (calculated - tick) / wheel.length;

                final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
                // 计算该任务对应的bucket索引值
                int stopIndex = (int) (ticks & mask);
                // 加入bucket
                HashedWheelBucket bucket = wheel[stopIndex];
                bucket.addTimeout(timeout);
            }
        }

最终,expireTimeouts(long deadline) 方法会执行那些需要在该轮次执行的所有任务,如果不是该伦次执行的话,就将其remainingRounds 属性值减一(代码如下)

  public void expireTimeouts(long deadline) {
            HashedWheelTimeout timeout = head;

            // process all timeouts
            while (timeout != null) {
                HashedWheelTimeout next = timeout.next;
                if (timeout.remainingRounds <= 0) {
                    next = remove(timeout);
                    if (timeout.deadline <= deadline) {
                        timeout.expire();
                    } else {
                        // The timeout was placed into a wrong slot. This should never happen.
                        throw new IllegalStateException(String.format(
                                "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
                    }
                } else if (timeout.isCancelled()) {
                    next = remove(timeout);
                } else {
                    timeout.remainingRounds --;
                }
                timeout = next;
            }
        }

应用

xxl-job中就是用了自己简易的时间轮,关于Netty中时间轮的应用,Redisson中分布式锁自动续费的看门狗,大家了解吗,这个看门狗的实现其实就是运用了Netty中的时间轮,可以自己先熟悉熟悉,后面有时间我们一起看一看他是如何来使用时间轮实现续费的功能的。

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 218,682评论 6 507
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 93,277评论 3 395
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 165,083评论 0 355
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 58,763评论 1 295
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 67,785评论 6 392
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 51,624评论 1 305
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 40,358评论 3 418
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 39,261评论 0 276
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 45,722评论 1 315
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,900评论 3 336
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 40,030评论 1 350
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 35,737评论 5 346
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 41,360评论 3 330
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,941评论 0 22
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,057评论 1 270
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 48,237评论 3 371
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 44,976评论 2 355

推荐阅读更多精彩内容

  • 背景 在实际的业务场景中,我们常常需要周期性执行一些任务,比如巡查系统资源,处理过期数据等等。这些事情如果人工去执...
    小波同学阅读 5,299评论 0 19
  • 前言 国庆节快到了,项目也结束了,最近一直在总结,接到了XXL-JOB调研任务。在“官方时间”摸鱼,笔者还是很开心...
    ZacPark阅读 859评论 0 0
  • 最近有朋友问到定时任务相关的问题。 于是,我简单写了一篇文章总结一下定时任务的一些概念以及一些常见的定时任务技术选...
    jeffrey_hjf阅读 2,747评论 0 0
  • 来自公众号:JavaGuide,作者Guide哥 最近有朋友问到定时任务相关的问题。 于是,我简单写了一篇文章总结...
    焌燈儿阅读 2,010评论 0 0
  • 近期花了一些时间翻了xxl-job的源码,稍作分析,希望能从如此成熟的框架中洞悉一些分布式任务调度的本质。 本文的...
    丑人林宗己阅读 1,341评论 0 2