任务调度之时间轮实现 | 京东云技术团队

前言

在生活中太阳的东升西落，鸟类的南飞北归，四级的轮换，每天的上下班，海水的潮汐，每月的房租车贷等等，如果用程序员的视角看，这就是一个个的定时任务，在日常的开发工作中也有很多的定时任务场景：

1. 数仓系统凌晨进行的数据同步
2. 订单12小时未支付的状态校验
3. rpc调用超时时间的校验
4. 缓存数据失效时间的延长
5. 定时开启的促销活动
6. ……

假如现在有一个任务需要3s后执行，你会如何实现？

简单点，直接一个线程的休眠，thread.sleep(3000),一行代码就能达到目的，但是性能嘛……，由于每个任务都需要一个单独的线程，当系统中存在大量任务，

任务调度

假如，现在有一个任务需要3s后执行，你会如何实现呢？

简单点，直接一个休眠，让线程sleep 3s，不就达到目的了吗？但是性能嘛……，由于每个任务都需要一个单独的线程，在系统中存在大量任务时，这种方案的消耗是极其巨大的，那么如何实现高效的调度呢？大佬们低头看了一眼手表，一个算法出现了

时间轮的数据结构

如图所示，这就是时间轮的一个基础结构，一个存储了定时任务的环形队列，可以理解为一个时间钟，队列的每个节点称为时间槽，每个槽位又使用列表存储着需要执行的定时任务。和生活中的钟表运行机制一样，每隔固定的单位时间，就会从一个槽位跳到下一个槽位，就像秒针跳动了一次，再取出当前槽位的任务进行执行。假如固定单位时间为1S，当前槽位位2，如果需要插入一个3S后的任务，就会在槽位5的的列表里加上当前任务。等指针运行到第五个槽位时，取出任务执行就可以了。

时间轮的最大优势是在时间复杂度上的优势，一个任务简单的生命周期：

1. 创建任务，插入到数据结构中。
2. 查询任务，找到满足条件的任务
3. 执行任务。
4. 任务归档，从任务调度的列表中删出。

其中第三步的执行时间是固定的，所以1，2，4这三部就的时间复杂度就决定了整个任务调度流程的复杂度，而时间轮是链式存储结构，所以在增删和查询时，时间复杂度都是0（1），其他常见的任务调度算法例如最小堆和红黑树以及跳表。

最小堆是一颗完全二叉树而且子节点的值总是大于等于父节点的值，所以在插入时候需要判断父节点的关系，它的时间添加操作时间复杂度是O(logn),在任务执行时，只需要判断最顶节点就行，所以它的查询时间复杂度时哦O（1）。

根据红黑树的特性已经被归纳法证明它的增加的时间复杂度是O（logn）,查找最小节点的时间复杂度位O（h）。

跳表的的本质是实现二分查找法的有序链表，但是他有多个层级，和红黑树的高度值相似，它的时间复杂度也是O（logn）

高级时间轮

如上图所示，如果一个刻度代表1S，那么一个周期就是1分钟，但是如果我一个任务是在3分钟后执行呢，如果是在一个12小时后执行呢？

当然如果是单纯的增加环形链表的长度也是可以的，直接扩大到3600*24，一天一个周期，直接放进来。但是还有更好的办法。

带轮次标记的任务

任务执行轮次的计算公式：（（任务执行时间-当前时间）/固定单位时间）%槽位数量

根据槽位计算公式可以算出当前任务需要插入执行的轮次，我在任务上面加一个字段round，当每次执行到该槽位时，就遍历该槽位的任务列表，每个任务的round-1，取出来round=0的任务执行就行。

1for(Task task:taskList){
2  int round= task.getRound();
3   round=(round-1);
4   task.setRound(round);
5   if(round==0){
6     doTask(task);
7   }
8}

如果任务间隔不是很大，看起来也是不错的一种解决方式。

但是工作中有很多任务，延迟执行的时间是很久以后的，例如延保履约服务成功之后会有一个7天自动完成的定时任务，甚至有一些几年后才会执行的任务，如果都用round来处理的话，那这个round将会变的非常大的一个数字，也会在任务列表中插入很多当前不需要执行的任务，如果每次都执行上面的逻辑，显然会浪费大量的资源。

多层时间轮

多层时间轮的核心思想是：

就上上图的水表，有很多小的表盘，但是每个表盘的刻度其实是不一样，又或者手表里的时分秒或者日历上的年月日。

针对时间复杂度的问题：不做遍历计算round,只要到了当前槽位，就把任务列表的所有任务拿出来执行。

针对空间复杂度的问题：分层，每个层级的时间轮刻度不一样，多个时间轮协调工作。

如上图所示，第一次时间轮，每个刻度是1ms,一轮是20ms,第二个层时间轮的刻度是20ms,一轮就是400ms,第三层的刻度是400ms,一轮就是8000ms,每层的周期就等于 20ms *2的n次方。这要使用多层级时间轮就可以很容易把任务区分开来。每当高层次时间轮到达当前节点，就把任务降级到低层级的时间轮上。对于400ms的时间轮来说，小于1ms和小于399ms的任务都是过期任务，只要不大于400ms，都认为是过期任务。

代码实现的话，往上也有很多，最近比较火热的POWER-JOB的分布式调度框架就是才有的时间轮算法，粘贴下核心代码大家看下：

1.首先定义了一个任务接口

1public interface TimerTask extends Runnable {
2}

2.调度中的任务对象

1public interface TimerFuture {
2
3    /**
4     * 获取实际要执行的任务
5     * @return
6     */
7    TimerTask getTask();
8
9    /**
10     * 取消任务
11     * @return
12     */
13    boolean cancel();
14
15    /**
16     * 任务是否取消
17     * @return
18     */
19    boolean isCancelled();
20
21    /**
22     * 任务是否完成
23     * @return
24     */
25    boolean isDone();
26}

3.调度器接口

1public interface Timer {
2
3    /**
4     * 调度定时任务
5     */
6    TimerFuture schedule(TimerTask task, long delay, TimeUnit unit);
7
8    /**
9     * 停止所有调度任务
10     */
11    Set<TimerTask> stop();
12}

4.时间轮的实现

1public class HashedWheelTimer implements Timer {
2
3    private final long tickDuration;
4    private final HashedWheelBucket[] wheel;
5    private final int mask;
6
7    private final Indicator indicator;
8
9    private final long startTime;
10
11    private final Queue<HashedWheelTimerFuture> waitingTasks = Queues.newLinkedBlockingQueue();
12    private final Queue<HashedWheelTimerFuture> canceledTasks = Queues.newLinkedBlockingQueue();
13
14    private final ExecutorService taskProcessPool;
15
16    public HashedWheelTimer(long tickDuration, int ticksPerWheel) {
17        this(tickDuration, ticksPerWheel, 0);
18    }
19
20    /**
21     * 新建时间轮定时器
22     * @param tickDuration 时间间隔，单位毫秒（ms）
23     * @param ticksPerWheel 轮盘个数
24     * @param processThreadNum 处理任务的线程个数，0代表不启用新线程（如果定时任务需要耗时操作，请启用线程池）
25     */
26    public HashedWheelTimer(long tickDuration, int ticksPerWheel, int processThreadNum) {
27
28        this.tickDuration = tickDuration;
29
30        // 初始化轮盘，大小格式化为2的N次，可以使用 & 代替取余
31        int ticksNum = CommonUtils.formatSize(ticksPerWheel);
32        wheel = new HashedWheelBucket[ticksNum];
33        for (int i = 0; i < ticksNum; i++) {
34            wheel[i] = new HashedWheelBucket();
35        }
36        mask = wheel.length - 1;
37
38        // 初始化执行线程池
39        if (processThreadNum <= 0) {
40            taskProcessPool = null;
41        }else {
42            ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("HashedWheelTimer-Executor-%d").build();
43            // 这里需要调整一下队列大小
44            BlockingQueue<Runnable> queue = Queues.newLinkedBlockingQueue(8192);
45            int core = Math.max(Runtime.getRuntime().availableProcessors(), processThreadNum);
46            // 基本都是 io 密集型任务
47            taskProcessPool = new ThreadPoolExecutor(core, 2 * core,
48                    60, TimeUnit.SECONDS,
49                    queue, threadFactory, RejectedExecutionHandlerFactory.newCallerRun("PowerJobTimeWheelPool"));
50        }
51
52        startTime = System.currentTimeMillis();
53
54        // 启动后台线程
55        indicator = new Indicator();
56        new Thread(indicator, "HashedWheelTimer-Indicator").start();
57    }
58
59    @Override
60    public TimerFuture schedule(TimerTask task, long delay, TimeUnit unit) {
61
62        long targetTime = System.currentTimeMillis() + unit.toMillis(delay);
63        HashedWheelTimerFuture timerFuture = new HashedWheelTimerFuture(task, targetTime);
64
65        // 直接运行到期、过期任务
66        if (delay <= 0) {
67            runTask(timerFuture);
68            return timerFuture;
69        }
70
71        // 写入阻塞队列，保证并发安全（性能进一步优化可以考虑 Netty 的 Multi-Producer-Single-Consumer队列）
72        waitingTasks.add(timerFuture);
73        return timerFuture;
74    }
75
76    @Override
77    public Set<TimerTask> stop() {
78        indicator.stop.set(true);
79        taskProcessPool.shutdown();
80        while (!taskProcessPool.isTerminated()) {
81            try {
82                Thread.sleep(100);
83            }catch (Exception ignore) {
84            }
85        }
86        return indicator.getUnprocessedTasks();
87    }
88
89    /**
90     * 包装 TimerTask，维护预期执行时间、总圈数等数据
91     */
92    private final class HashedWheelTimerFuture implements TimerFuture {
93
94        // 预期执行时间
95        private final long targetTime;
96        private final TimerTask timerTask;
97
98        // 所属的时间格，用于快速删除该任务
99        private HashedWheelBucket bucket;
100        // 总圈数
101        private long totalTicks;
102        // 当前状态 0 - 初始化等待中，1 - 运行中，2 - 完成，3 - 已取消
103        private int status;
104
105        // 状态枚举值
106        private static final int WAITING = 0;
107        private static final int RUNNING = 1;
108        private static final int FINISHED = 2;
109        private static final int CANCELED = 3;
110
111        public HashedWheelTimerFuture(TimerTask timerTask, long targetTime) {
112
113            this.targetTime = targetTime;
114            this.timerTask = timerTask;
115            this.status = WAITING;
116        }
117
118        @Override
119        public TimerTask getTask() {
120            return timerTask;
121        }
122
123        @Override
124        public boolean cancel() {
125            if (status == WAITING) {
126                status = CANCELED;
127                canceledTasks.add(this);
128                return true;
129            }
130            return false;
131        }
132
133        @Override
134        public boolean isCancelled() {
135            return status == CANCELED;
136        }
137
138        @Override
139        public boolean isDone() {
140            return status == FINISHED;
141        }
142    }
143
144    /**
145     * 时间格（本质就是链表，维护了这个时刻可能需要执行的所有任务）
146     */
147    private final class HashedWheelBucket extends LinkedList<HashedWheelTimerFuture> {
148
149        public void expireTimerTasks(long currentTick) {
150
151            removeIf(timerFuture -> {
152
153                // processCanceledTasks 后外部操作取消任务会导致 BUCKET 中仍存在 CANCELED 任务的情况
154                if (timerFuture.status == HashedWheelTimerFuture.CANCELED) {
155                    return true;
156                }
157
158                if (timerFuture.status != HashedWheelTimerFuture.WAITING) {
159                    log.warn("[HashedWheelTimer] impossible, please fix the bug");
160                    return true;
161                }
162
163                // 本轮直接调度
164                if (timerFuture.totalTicks <= currentTick) {
165
166                    if (timerFuture.totalTicks < currentTick) {
167                        log.warn("[HashedWheelTimer] timerFuture.totalTicks < currentTick, please fix the bug");
168                    }
169
170                    try {
171                        // 提交执行
172                        runTask(timerFuture);
173                    }catch (Exception ignore) {
174                    } finally {
175                        timerFuture.status = HashedWheelTimerFuture.FINISHED;
176                    }
177                    return true;
178                }
179
180                return false;
181            });
182
183        }
184    }
185
186    private void runTask(HashedWheelTimerFuture timerFuture) {
187        timerFuture.status = HashedWheelTimerFuture.RUNNING;
188        if (taskProcessPool == null) {
189            timerFuture.timerTask.run();
190        }else {
191            taskProcessPool.submit(timerFuture.timerTask);
192        }
193    }
194
195    /**
196     * 模拟时针转动的线程
197     */
198    private class Indicator implements Runnable {
199
200        private long tick = 0;
201
202        private final AtomicBoolean stop = new AtomicBoolean(false);
203        private final CountDownLatch latch = new CountDownLatch(1);
204
205        @Override
206        public void run() {
207
208            while (!stop.get()) {
209
210                // 1. 将任务从队列推入时间轮
211                pushTaskToBucket();
212                // 2. 处理取消的任务
213                processCanceledTasks();
214                // 3. 等待指针跳向下一刻
215                tickTack();
216                // 4. 执行定时任务
217                int currentIndex = (int) (tick & mask);
218                HashedWheelBucket bucket = wheel[currentIndex];
219                bucket.expireTimerTasks(tick);
220
221                tick ++;
222            }
223            latch.countDown();
224        }
225
226        /**
227         * 模拟指针转动，当返回时指针已经转到了下一个刻度
228         */
229        private void tickTack() {
230
231            // 下一次调度的绝对时间
232            long nextTime = startTime + (tick + 1) * tickDuration;
233            long sleepTime = nextTime - System.currentTimeMillis();
234
235            if (sleepTime > 0) {
236                try {
237                    Thread.sleep(sleepTime);
238                }catch (Exception ignore) {
239                }
240            }
241        }
242
243        /**
244         * 处理被取消的任务
245         */
246        private void processCanceledTasks() {
247            while (true) {
248                HashedWheelTimerFuture canceledTask = canceledTasks.poll();
249                if (canceledTask == null) {
250                    return;
251                }
252                // 从链表中删除该任务（bucket为null说明还没被正式推入时间格中，不需要处理）
253                if (canceledTask.bucket != null) {
254                    canceledTask.bucket.remove(canceledTask);
255                }
256            }
257        }
258
259        /**
260         * 将队列中的任务推入时间轮中
261         */
262        private void pushTaskToBucket() {
263
264            while (true) {
265                HashedWheelTimerFuture timerTask = waitingTasks.poll();
266                if (timerTask == null) {
267                    return;
268                }
269
270                // 总共的偏移量
271                long offset = timerTask.targetTime - startTime;
272                // 总共需要走的指针步数
273                timerTask.totalTicks = offset / tickDuration;
274                // 取余计算 bucket index
275                int index = (int) (timerTask.totalTicks & mask);
276                HashedWheelBucket bucket = wheel[index];
277
278                // TimerTask 维护 Bucket 引用，用于删除该任务
279                timerTask.bucket = bucket;
280
281                if (timerTask.status == HashedWheelTimerFuture.WAITING) {
282                    bucket.add(timerTask);
283                }
284            }
285        }
286
287        public Set<TimerTask> getUnprocessedTasks() {
288            try {
289                latch.await();
290            }catch (Exception ignore) {
291            }
292
293            Set<TimerTask> tasks = Sets.newHashSet();
294
295            Consumer<HashedWheelTimerFuture> consumer = timerFuture -> {
296                if (timerFuture.status == HashedWheelTimerFuture.WAITING) {
297                    tasks.add(timerFuture.timerTask);
298                }
299            };
300
301            waitingTasks.forEach(consumer);
302            for (HashedWheelBucket bucket : wheel) {
303                bucket.forEach(consumer);
304            }
305            return tasks;
306        }
307    }
308}

作者：京东保险 陈建华

来源：京东云开发者社区