前言

前段时间在给自己的玩具项目设计的时候就遇到了一个场景需要定时任务，于是就趁机了解了目前主流的一些定时任务方案，比如下面这些：

  • Timer（halo 博客源码中用到了）
  • ScheduledExecutorService
  • ThreadPoolTaskScheduler（基于 ScheduledExecutorService）
  • Netty 的 schedule（用到了 PriorityQueue）
  • Netty 的 HashedWheelTimer（时间轮）
  • Kafka 的 TimingWheel（层级时间轮）

还有一些分布式的定时任务：

  • Quartz
  • xxl-job（我实习公司就在用这个）
  • …

因为我玩具项目实现业务 ACK 的方案就打算用 HashedWheelTimer，所以本节核心是分析 HashedWheelTimer，另外会提下它与 schedule 的区别，其它定时任务实现原理就请自动 Google 吧。

Netty Version：4.1.42

HashedWheelTimer 实现图示

HashedWheelTimer实现图示.png

大致有个理解就行，关于蓝色格子中的数字，其实就是剩余时钟轮数，这里听不懂也没关系，等后面看到源码解释就懂了（大概）。

HashedWheelTimer 简答使用例子

这里顺便列出 schedule 的使用方式，下面是某个 Handler 中的代码：

  1. @Override
  2. public void handlerAdded(final ChannelHandlerContext ctx) {
  3.     // 定时任务
  4.     ScheduledFuture<?> hello_world = ctx.executor().schedule(() -> {
  5.         ctx.channel().write("hello world");
  6.     }, 3, TimeUnit.SECONDS);
  8.     // 构造一个 Timer 实例，同样只执行一次
  9.     Timer timer = new HashedWheelTimer();
  10.     Timeout timeout1 = timer.newTimeout(timeout -> System.out.println("5s 后执行该任务"), 5, TimeUnit.SECONDS);
  12.     // 取消任务
  13.     timeout1.cancel();
  14. }

HashedWheelTimer 源码

继承关系、方法

继承关系&方法.png

构造函数、属性

请记住这些属性的是干啥用的，后面会频繁遇到： io.netty.util.HashedWheelTimer#HashedWheelTimer(java.util.concurrent.ThreadFactory, long, java.util.concurrent.TimeUnit, int, boolean, long)

  1. public HashedWheelTimer(
  2.         ThreadFactory threadFactory,
  3.         long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection,
  4.         long maxPendingTimeouts) {
  6.     if (threadFactory == null) {
  7.         throw new NullPointerException("threadFactory");
  8.     }
  9.     if (unit == null) {
  10.         throw new NullPointerException("unit");
  11.     }
  12.     if (tickDuration <= 0) {
  13.         throw new IllegalArgumentException("tickDuration must be greater than 0: " + tickDuration);
  14.     }
  15.     if (ticksPerWheel <= 0) {
  16.         throw new IllegalArgumentException("ticksPerWheel must be greater than 0: " + ticksPerWheel);
  17.     }
  19.     // 初始化时间轮，数组长度必须是2的幂次方，便于取模
  20.     // Normalize ticksPerWheel to power of two and initialize the wheel.
  21.     wheel = createWheel(ticksPerWheel);
  22.     // 用于取模运算, tick & mask
  23.     mask = wheel.length - 1;
  25.     // Convert tickDuration to nanos.
  26.     // 毫秒转纳秒
  27.     long duration = unit.toNanos(tickDuration);
  29.     // Prevent overflow.
  30.     // 防止溢出
  31.     if (duration >= Long.MAX_VALUE / wheel.length) {
  32.         throw new IllegalArgumentException(String.format(
  33.                 "tickDuration: %d (expected: 0 < tickDuration in nanos < %d",
  34.                 tickDuration, Long.MAX_VALUE / wheel.length));
  35.     }
  37.     // 时间刻度设置太小，自动设置为MILLISECOND_NANOS， 并弹出警告日志
  38.     if (duration < MILLISECOND_NANOS) {
  39.         logger.warn("Configured tickDuration {} smaller then {}, using 1ms.",
  40.                     tickDuration, MILLISECOND_NANOS);
  41.         this.tickDuration = MILLISECOND_NANOS;
  42.     } else {
  43.         this.tickDuration = duration;
  44.     }
  46.     // 初始化工作线程，注意这里还没有启动
  47.     // 另外需要注意的是本类中的startTime是在worker第一次启动之后才初始化的，
  48.     // 跟io.netty.util.concurrent.ScheduledFutureTask.START_TIME在类加载的时候初始化是不一样的
  49.     workerThread = threadFactory.newThread(worker);
  51.     // 用来跟踪内存问题的，本节忽略，主讲定时任务的实现
  52.     leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;
  54.     // 最大允许任务等待数
  55.     this.maxPendingTimeouts = maxPendingTimeouts;
  57.     // HashedWheelTimer实例如果超过64个就会弹出警告，告诉你HashedWheelTimer不是这样用的，单个应用只需要一个单例即可
  58.     if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
  59.         WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
  60.         reportTooManyInstances();
  61.     }
  62. }

添加定时任务

添加定时任务其实就是 Timer 接口的 newTimeOut 方法： io.netty.util.HashedWheelTimer#newTimeout

  1. @Override
  2. public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
  3.     if (task == null) {
  4.         throw new NullPointerException("task");
  5.     }
  6.     if (unit == null) {
  7.         throw new NullPointerException("unit");
  8.     }
  10.     // 获取当前等待任务数
  11.     long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();
  13.     // 如果超出最大等待
  14.     if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
  15.         pendingTimeouts.decrementAndGet();
  16.         throw new RejectedExecutionException("Number of pending timeouts ("
  17.             + pendingTimeoutsCount + ") is greater than or equal to maximum allowed pending "
  18.             + "timeouts (" + maxPendingTimeouts + ")");
  19.     }
  21.     // 尝试启动workerThread，startTime=0时会startTimeInitialized.await()，最终就是调用Worker的run方法
  22.     start();
  24.     // Add the timeout to the timeout queue which will be processed on the next tick.
  25.     // During processing all the queued HashedWheelTimeouts will be added to the correct HashedWheelBucket.
  26.     // 这条算式我们可以稍微改下，更容易理解些：
  27.     // long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
  28.     //                                 ↓
  29.     // long deadline = unit.toNanos(delay) - (System.nanoTime() - startTime)
  30.     // 我感觉这样更容易理解些，含义为： 距离任务执行的剩余时间 = 任务截止时间 - (当前时间 - 任务对象初始化时间)
  31.     long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
  33.     // Guard against overflow.
  34.     if (delay > 0 && deadline < 0) {
  35.         deadline = Long.MAX_VALUE;
  36.     }
  38.     // 构建任务对象
  39.     HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
  40.     // 将任务对象添加到mpsc队列中，mpsc是多生产者单消费者的队列模型，另外mpscQueue是无锁队列，靠的CAS实现的。
  41.     timeouts.add(timeout);
  42.     // 返回任务对象，该对象可以用于取消任务、获取任务信息等
  43.     return timeout;
  44. }

这里我们再跟进 start 方法看看： io.netty.util.HashedWheelTimer#start

  1. public void start() {
  2.     switch (WORKER_STATE_UPDATER.get(this)) {
  3.         case WORKER_STATE_INIT:
  4.             if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
  5.                 // 一半会来到这里，最终就是调用到Worker的run方法
  6.                 workerThread.start();
  7.             }
  8.             break;
  9.         case WORKER_STATE_STARTED:
  10.             break;
  11.         case WORKER_STATE_SHUTDOWN:
  12.             throw new IllegalStateException("cannot be started once stopped");
  13.         default:
  14.             throw new Error("Invalid WorkerState");
  15.     }
  17.     // Wait until the startTime is initialized by the worker.
  18.     while (startTime == 0) {
  19.         try {
  20.             // 如果startTime异常，Worker的run方法会处理这种异常，重新唤醒
  21.             startTimeInitialized.await();
  22.         } catch (InterruptedException ignore) {
  23.             // Ignore - it will be ready very soon.
  24.         }
  25.     }
  26. }

定时任务执行

定时任务的执行逻辑其实就在 Worker 的 run 方法中： io.netty.util.HashedWheelTimer.Worker#run

  1. // 用于处理取消的任务
  2. private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
  4. // 时钟指针转动的次数
  5. private long tick;
  7. @Override
  8. public void run() {
  9.     // Initialize the startTime.
  10.     startTime = System.nanoTime();
  11.     if (startTime == 0) {
  12.         // We use 0 as an indicator for the uninitialized value here, so make sure it's not 0 when initialized.
  13.         startTime = 1;
  14.     }
  16.     // Notify the other threads waiting for the initialization at start().
  17.     // 之前如果startTime=0，就会进入await状态，这里就要唤醒它
  18.     startTimeInitialized.countDown();
  20.     do {
  21.         /*
  22.             * 等待到下一次 tick 时如果没有时间延迟返回tickDuration * (tick + 1);
  23.             * 如果延迟了则不空转，立马返回“当前时间”
  24.             * 这个“当前时间”是什么呢？比如时钟指针原本第三次 tick 是在300ms，但是由于前面的任务阻塞了50ms，导致进来的时候已经是350ms了
  25.             * 那么这里的返回值就会变成350ms，至于返回值变成350ms会怎么样？貌似也没有怎么样，就是不等待马上执行罢了
  26.             */
  27.         final long deadline = waitForNextTick();
  28.         if (deadline > 0) {
  29.             // 与运算取模，取出数组桶的坐标，相信这个没少见过了
  30.             int idx = (int) (tick & mask);
  31.             // 前面说过HashedWheelTimeout是可以取消任务的，其实就是在这里取消的
  32.             processCancelledTasks();
  33.             // 在时间轮中取出“指针指向的块”
  34.             HashedWheelBucket bucket =
  35.                     wheel[idx];
  36.             // 将任务填充到时间块中
  37.             transferTimeoutsToBuckets();
  38.             // 取出任务并执行
  39.             bucket.expireTimeouts(deadline);
  40.             tick++;
  41.         }
  42.     } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
  44.     // Fill the unprocessedTimeouts so we can return them from stop() method.
  45.     for (HashedWheelBucket bucket: wheel) {
  46.         bucket.clearTimeouts(unprocessedTimeouts);
  47.     }
  48.     for (;;) {
  49.         HashedWheelTimeout timeout = timeouts.poll();
  50.         if (timeout == null) {
  51.             break;
  52.         }
  53.         if (!timeout.isCancelled()) {
  54.             unprocessedTimeouts.add(timeout);
  55.         }
  56.     }
  57.     // 处理取消的任务
  58.     processCancelledTasks();
  59. }
  • 取消任务的逻辑这里就不展开看了，也比较简单，有兴趣自行补充即可。

看看上面的 transferTimeoutsToBuckets 方法，如果你看不懂上面图中蓝色格子数字是什么意思，那就认真看看这个方法： io.netty.util.HashedWheelTimer.Worker#transferTimeoutsToBuckets

  1. private void transferTimeoutsToBuckets() {
  2.     // transfer only max. 100000 timeouts per tick to prevent a thread to stale the workerThread when it just
  3.     // adds new timeouts in a loop.
  4.     for (int i = 0; i < 100000; i++) {
  5.         // 取出一个任务对象
  6.         HashedWheelTimeout timeout = timeouts.poll();
  7.         if (timeout == null) {
  8.             // all processed
  9.             break;
  10.         }
  11.         // 如果任务被取消了，则直接过掉
  12.         if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
  13.             // Was cancelled in the meantime.
  14.             continue;
  15.         }
  17.         /*
  18.             * remainingRounds的含义就是:时钟还要完整转几回才能执行到任务
  19.             * 比如你的任务是在2500ms之后才执行的（deadline = 2500ms），时钟总共10个刻度，而 tickDuration 为100ms，当前时钟指针已经拨动三次（tick=3）
  20.             * 那 2500 / 100 = 25
  21.             * (25 - 3) / 10 约等于 2
  22.             * 2 就表示 时钟转完当前圈（25-10=15），还要再转一圈（15-10），在第三圈才能执行到该任务
  23.             */
  24.         long calculated = timeout.deadline / tickDuration;
  25.         timeout.remainingRounds = (calculated - tick) / wheel.length;
  27.         final long ticks = Math.max(calculated, tick); // Ensure we don't schedule for past.
  28.         int stopIndex = (int) (ticks & mask);
  30.         // 将任务填充到“时间块”中
  31.         HashedWheelBucket bucket = wheel[stopIndex];
  32.         bucket.addTimeout(timeout);
  33.     }
  34. }

继续看看上面 run 方法中的 bucket.expireTimeouts(deadline);，这里面就是拿出任务并执行的逻辑： io.netty.util.HashedWheelTimer.HashedWheelBucket#expireTimeouts

  1. /**
  2.     * Expire all {@link HashedWheelTimeout}s for the given {@code deadline}.
  3.     */
  4. public void expireTimeouts(long deadline) {
  5.     HashedWheelTimeout timeout = head;
  7.     // process all timeouts
  8.     while (timeout != null) {
  9.         HashedWheelTimeout next = timeout.next;
  10.         // 如果剩余轮数 <=0，则表示当前轮就要执行任务了
  11.         if (timeout.remainingRounds <= 0) {
  12.             next = remove(timeout);
  13.             if (timeout.deadline <= deadline) {
  14.                 // 执行任务
  15.                 timeout.expire();
  16.             } else {
  17.                 // The timeout was placed into a wrong slot. This should never happen.
  18.                 throw new IllegalStateException(String.format(
  19.                         "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
  20.             }
  21.         }
  22.         // 如果任务被取消了
  23.         else if (timeout.isCancelled()) {
  24.             next = remove(timeout);
  25.         }
  26.         // 如果任务没被取消，而且剩余轮数>0，则扣除轮数，让任务继续等到至后面轮数
  27.         else {
  28.             timeout.remainingRounds --;
  29.         }
  30.         timeout = next;
  31.     }
  32. }

和 schedule 对比

关于 schedule 方法加入的定时任务什么时候被执行，你可以参考我之前写的这篇博客，在时间操作上和 HashedWheelTimer 大同小异。

schedule 方法也是 Netty 的定时任务实现之一，但是底层的数据结构和 HashedWheelTimer 不一样，schedule 方法用到的数据结构其实和 ScheduledExecutorService 类似，是 PriorityQueue，它是一个优先级的队列。

除此之外，schedule 方法其实也用到 MpscQueue，只是任务执行的时候，会把任务从 PriorityQueue 转移到 MpscQueue 上。

下面来跟踪下 schedule 方法看看，由于主要是看数据结构的区别，所以一些地方在这里我就不深追了

首先来到如下代码： io.netty.util.concurrent.AbstractScheduledEventExecutor#schedule(java.lang.Runnable, long, java.util.concurrent.TimeUnit)

  1. @Override
  2. public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
  3.     ObjectUtil.checkNotNull(command, "command");
  4.     ObjectUtil.checkNotNull(unit, "unit");
  5.     if (delay < 0) {
  6.         delay = 0;
  7.     }
  8.     validateScheduled0(delay, unit);
  10.     return schedule(new ScheduledFutureTask<Void>(
  11.             this, command, null, ScheduledFutureTask.deadlineNanos(unit.toNanos(delay))));
  12. }

继续跟进 schedule 方法看看： io.netty.util.concurrent.AbstractScheduledEventExecutor#schedule(io.netty.util.concurrent.ScheduledFutureTask<V>)

  1. private <V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {
  2.     if (inEventLoop()) {
  3.         scheduledTaskQueue().add(task.setId(nextTaskId++));
  4.     } else {
  5.         executeScheduledRunnable(new Runnable() {
  6.             @Override
  7.             public void run() {
  8.                 scheduledTaskQueue().add(task.setId(nextTaskId++));
  9.             }
  10.         }, true, task.deadlineNanos());
  11.     }
  13.     return task;
  14. }

继续跟进 scheduledTaskQueue()方法： io.netty.util.concurrent.AbstractScheduledEventExecutor#scheduledTaskQueue

  1. PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue() {
  2.     if (scheduledTaskQueue == null) {
  3.         scheduledTaskQueue = new DefaultPriorityQueue<ScheduledFutureTask<?>>(
  4.                 SCHEDULED_FUTURE_TASK_COMPARATOR,
  5.                 // Use same initial capacity as java.util.PriorityQueue
  6.                 11);
  7.     }
  8.     return scheduledTaskQueue;
  9. }

可以看到返回值就是 PriorityQueue，它是一个最小堆实现的优先队列。

扩展

不同实现的时间复杂度

这里我就直接贴下网上大佬给出的解释：

如果使用最小堆实现的优先级队列： 最小堆.png

  • 大致意思就是你的任务如果插入到堆顶，时间复杂度为 O(log(n))。

如果使用链表（既然有说道，那就扩展下）： 链表.png

  • 中间插入后的事件复杂度为 O(n)

单个时间轮： 单个时间轮.png

  • 复杂度可以降至 O(1)。

记录轮数的时间轮（其实就是文章开头的那个）： 记录轮数的时间轮.png

层级时间轮： 层级时间轮.png

  • 时间复杂度是 O(n)，n 是轮子的数量，除此之外还要计算一个轮子上的 bucket。

单时间轮缺点

根据上面的图其实不难理解，如果任务是很久之后才执行的、同时要保证任务低延迟，那么单个时间轮所需的 bucket 数就会变得非常多，从而导致内存占用持续升高（CPU 空转时间还是不变的，仅仅是内存需求变高了），如下图：

image.png

Netty 对于单个时间轮的优化方式就是记录下 remainingRounds，从而减少 bucket 过多的内存占用。

时间轮和 PriorityQueue 对比

看完上面的时间复杂度对比，你可能会觉得：

  • Q：时间轮的复杂度只有 O(1)，schedule 和 ScheduledExecutorService 这种都是 O(log(n))，那时间轮不是碾压吗？

  • A：你不要忘了，如果任务是在很久之后才执行的，那么时间轮就会产生很多空转，这是非常浪费 CPU 性能的，这种空转消耗可以通过增大 tickDuration 来避免，但这样做又会产生降低定时任务的精度，可能导致一些任务推到很迟才执行。

  • A：而 ScheduledExecutorService 不会有这个问题。

另外，Netty 时间轮的实现模型抽象出来是大概这个样子的：

  1. for(Tasks task : tasks) {
  2.     task.doXxx();
  3. }

这个抽象是个什么意思呢？你要注意一个点，这里的任务循环执行是同步的，这意味着你第一个任务执行很慢延迟很高，那么后面的任务全都会被堵住，所以你加进时间轮的任务不可以是耗时任务，比如一些延迟很高的数据库查询，如果有这种耗时任务，最好再嵌入线程池处理，不要让任务阻塞在这一层。

原文链接：https://wenjie.store/archives/netty-hashedwheeltimer-and-schedule