Netty源码—10.Netty工具之时间轮

1.什么是时间轮


简单来说，时间轮是一个高效使用线程资源进行批量化调度的调度器。首先把大批量的调度使命全部绑定到同一个调度器上，然后使用这个调度器对全部使命进行管理、触发、以及运行，所以时间轮能高效管理各种延时使命、周期使命、通知使命。

时间轮是以时间作为刻度构成的一个环形队列，所以叫做时间轮。这个环形队列通过一个HashedWheelBucket[]数组来实现，数组的每个元素称为槽，每个槽可以存放一个定时使命列表，叫HashedWheelBucket。HashedWheelBucket是一个双向链表，链表的每个节点表示一个定时使命项HashedWheelTimeout。在HashedWheelTimeout中封装了真正的定时使命TimerTask。

时间轮由多个时间格构成，每个时间格代表当前时间轮的基本时间跨度ticketDuration，此中时间轮的时间格的个数是固定的。

如下图示，有16个时间格(槽)，假设每个时间格的单位是1s，那么整个时间轮走完一圈需要16s。每秒钟(即时间格的单位也可以为1ms、1min、1h等)指针会沿着顺时针方向转动一格。通过指针移动来获得每个时间格中的使命列表，然后遍历这个时间格内的双向链表的每个使命并实行，依此循环。

2.HashedWheelTimer是什么


Netty的HashedWheelTimer是一个大抵的定时器实现，之所以称为大抵的实现是由于该时间轮并没有严格定时地实行定时使命，而是在每隔一个时间隔断之后的时间节点实行，并实行当前时间节点之前到期的定时使命。

不外具体的定时使命的时间实行精度，可以通过调治HashedWheelTimer构造方法的时间隔断的大小来进行调治。在大多数网络应用的情况下，由于IO延迟的存在，所以并不会严格要求具体的时间实行精度。因此默认100ms的时间隔断可以满意大多数情况，不需要再花精神去调治该时间精度。

3.HashedWheelTimer的使用

1. public class HashedWheelTimerTest {
2.     //构建HashedWheelTimer时间轮
3.     //最后通过HASHED_WHEEL_TIMER.newTimeout()方法把需要延迟执行的任务添加到时间轮中
4.     private static final HashedWheelTimer HASHED_WHEEL_TIMER = new HashedWheelTimer(
5.         new DefaultThreadFactory("demo-timer"),//threadFactory参数表示创建处理任务的线程工厂
6.         100,//tickDuration参数表示每个时间格代表当前时间轮的基本时间跨度，这里是100ms，也就是指针100ms跳动一次，每次跳动一个窗格
7.         TimeUnit.MILLISECONDS,
8.         512,//ticksPerWheel参数表示时间轮上一共有多少个时间格，分配的时间格越多，占用内存空间就越大，这里是512
9.         true//leakDetection参数表示是否开启内存泄漏检测
10.     );
12.     public static void main(String[] args) {
13.         System.out.println("延时任务提交");
14.         //延时多久执行
15.         long delay = 10L;
16.         HASHED_WHEEL_TIMER.newTimeout(new TimerTask() {
17.             @Override
18.             public void run(Timeout timeout) throws Exception {
19.                 System.out.println("延时任务触发");
20.             }
21.         }, delay, TimeUnit.SECONDS);
22.     }
23. }

复制代码

4.HashedWheelTimer的运行流程


步调一：初始化时间轮
步调二：启动时间轮
步调三：添加使命，保存到延时使命队列
步调四：时间轮指针休眠阻塞，实现转动
步调五：休眠结束，指针指向下一个时间格(槽)
步调六：将已经取消的使命从对应的槽中移除
步调七：将延时使命队列的使命添加到对应的槽中
步调八：实行时间轮指针指向当前槽的到期使命

5.HashedWheelTimer中的关键字段


字段一：wheel
wheel是一个HashedWheelBucket数组，默认的数组大小是512。可以认为wheel是一个TimerTask的哈希表，它的哈希函数是使命的截止日期。所以每个时间轮的时间格数ticksPerWheel默认是512。

字段二：tickDuration
时间格跨度，默认100ms。

字段三：ticksPerWheel
时间轮的格子数，默认512。

字段四：maxPendingTimeouts
时间轮中使命的最大数目。

字段五：deadline
延时使命的截止时间，值为当前时间 + 延时使命的延时时间 - 时间轮启动时间。

字段六：tick
时间轮启动以来指针总的转动次数。

字段七：remainingRounds
槽中延时使命剩余的圈(轮)数，为0时则表示需要实行延时使命了。

6.HashedWheelTimer的构造方法


步调一：构造参数校验及给现实实行延时使命的线程池taskExecutor赋值

步调二：将ticksPerWheel(时间轮上的时间格数)向上取值为2的幂

步调三：初始化HashedWheelBucket数组wheel

步调四：校验tickDuration和ticksPerWheel

步调五：创建工作线程workerThread，用于指针转动和触发实行时间格里的延时使命

步调六：给时间轮中延时使命的最大数目maxPendingTimeouts赋值

步调七：查抄HashedWheelTimer的实例数目，如果大于64则打印error日志

1. //4.1.73.Final
2. public class HashedWheelTimer implements Timer {
3.     private final HashedWheelBucket[] wheel;
4.     private final int mask;
5.     private final long tickDuration;
6.     private final Thread workerThread;
7.     private final ResourceLeakTracker<HashedWheelTimer> leak;
8.     private final Worker worker = new Worker();
9.     private final long maxPendingTimeouts;
10.     private final Executor taskExecutor;
11.     ...
12.     //Creates a new timer.
13.     //@param threadFactory        创建线程的工厂
14.     //@param tickDuration         每格的时间间隔，默认100ms，0.1秒
15.     //@param unit                 时间单位，默认为毫秒
16.     //@param ticksPerWheel        时间轮的格子数，默认为512；如果传入的不是2的N次方，则会调整为大于等于该参数的第一个2的N次方，好处是可以优化hash值的计算
17.     //@param leakDetection        如果false，那么只有工作线程不是后台线程时才会追踪资源泄露，这个参数可以忽略
18.     //@param maxPendingTimeouts   最大的pending数量（时间轮中任务的最大数量），超过这个值之后调用将抛出异常，0或者负数表示没有限制，默认为-1
19.     //@param taskExecutor         任务线程池，用于执行提交的任务，调用者负责在不需要时关闭它
20.     //@throws NullPointerException     if either of threadFactory and unit is null
21.     //@throws IllegalArgumentException if either of tickDuration and ticksPerWheel is <= 0
22.     public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection, long maxPendingTimeouts, Executor taskExecutor) {
23.         //1.构造参数校验及给实际执行延时任务的线程池taskExecutor赋值
24.         checkNotNull(threadFactory, "threadFactory");
25.         checkNotNull(unit, "unit");
26.         checkPositive(tickDuration, "tickDuration");
27.         checkPositive(ticksPerWheel, "ticksPerWheel");
28.         this.taskExecutor = checkNotNull(taskExecutor, "taskExecutor");
30.         //2.将ticksPerWheel(时间轮上的时间格数)向上取值为2的幂，方便进行求商和取余计算
31.         //3.初始化时间轮wheel
32.         wheel = createWheel(ticksPerWheel);
33.         //mask的设计和HashMap一样，通过限制数组的大小为2的幂，利用位运算来替代取模运算，提高性能
34.         mask = wheel.length - 1;
35.       
36.         //4.校验tickDuration和ticksPerWheel
37.         //Convert tickDuration to nanos.
38.         long duration = unit.toNanos(tickDuration);
39.         //防止溢出
40.         //tickDuration * ticksPerWheel必须小于Long.MAX_VALUE
41.         if (duration >= Long.MAX_VALUE / wheel.length) {
42.             throw new IllegalArgumentException(String.format("tickDuration: %d (expected: 0 < tickDuration in nanos < %d", tickDuration, Long.MAX_VALUE / wheel.length));
43.         }
44.         //tickDuration不能小于1ms
45.         if (duration < MILLISECOND_NANOS) {
46.             logger.warn("Configured tickDuration {} smaller then {}, using 1ms.", tickDuration, MILLISECOND_NANOS);
47.             this.tickDuration = MILLISECOND_NANOS;
48.         } else {
49.             this.tickDuration = duration;
50.         }
51.         //5.创建工作线程，用于指针转动和触发时间格里的延时任务的执行
52.         workerThread = threadFactory.newThread(worker);
54.         leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;
55.         //6.给时间轮中任务的最大数量maxPendingTimeouts赋值
56.         this.maxPendingTimeouts = maxPendingTimeouts;
57.         //7.检查HashedWheelTimer的实例数量，如果大于64则打印error日志
58.         if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
59.             WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
60.             reportTooManyInstances();
61.         }
62.     }
63.    
64.     //初始化时间轮环形数组
65.     //@param ticksPerWheel
66.     private static HashedWheelBucket[] createWheel(int ticksPerWheel) {
67.         //ticksPerWheel不能大于2^30
68.         checkInRange(ticksPerWheel, 1, 1073741824, "ticksPerWheel");
69.         //将ticksPerWheel（轮子上的时间格数）向上取值为2的次幂
70.         ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
71.         //创建时间轮环形数组
72.         HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
73.         for (int i = 0; i < wheel.length; i ++) {
74.             wheel[i] = new HashedWheelBucket();
75.         }
76.         return wheel;
77.     }
78.     ...
79. }

复制代码

7.HashedWheelTimer添加使命和实行使命


(1)添加延时使命


步调一：将需要实行的延时使命数pendingTimeouts+1

步调二：如果pendingTimeouts超过maxPendingTimeouts，则抛出异常

步调三：启动工作线程，也就是启动时间轮

步调四：计算被添加的延时使命的截止时间=当前时间+当前使命实行的延迟时间-时间轮启动的时间

步调五：创建延时使命实例HashedWheelTimeout

步调六：将延时使命实例添加到延时使命队列timeouts中

注意：添加时会将延时使命添加到延时使命队列timeouts中。这个延时使命队列timeouts将会在下一个滴答声中进行处理惩罚(指针的下一次转动)。在处理惩罚过程中，全部排队的HashedWheelTimeout将被添加到正确的HashedWheelBucket。

1. public class HashedWheelTimer implements Timer {
2.     private final AtomicLong pendingTimeouts = new AtomicLong(0);//需要执行的延时任务数
3.     private final long maxPendingTimeouts;
4.     private volatile long startTime;
5.     private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
6.     private final Queue<HashedWheelTimeout> timeouts = PlatformDependent.newMpscQueue();//延时任务队列
7.     ...
8.     //添加延时任务
9.     //@param task 任务
10.     //@param delay 延时时间
11.     //@param unit 延时时间单位
12.     @Override
13.     public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
14.         checkNotNull(task, "task");
15.         checkNotNull(unit, "unit");
16.         //1.将需要执行的延时任务数pendingTimeouts + 1
17.         long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();
18.       
19.         //2.如果pendingTimeouts超过maxPendingTimeouts，则抛出异常
20.         if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
21.             pendingTimeouts.decrementAndGet();
22.             throw new RejectedExecutionException("Number of pending timeouts (" + pendingTimeoutsCount
23.                 + ") is greater than or equal to maximum allowed pending " + "timeouts (" + maxPendingTimeouts + ")");
24.         }
25.       
26.         //3.启动工作线程，即启动时间轮
27.         start();
28.      
29.         //将延时任务添加到延时任务队列timeouts中，该队列将在下一个滴答声中处理(指针的下一次转动)
30.         //在处理过程中，所有排队的HashedWheelTimeout将被添加到正确的HashedWheelBucket
32.         //4.计算任务的截止时间deadline = 当前时间 + 当前任务执行的延迟时间 - 时间轮启动的时间
33.         long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
34.         if (delay > 0 && deadline < 0) {
35.             deadline = Long.MAX_VALUE;
36.         }
37.       
38.         //5.创建延时任务实例HashedWheelTimeout
39.         HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
40.       
41.         //6.将延时任务实例添加到延时任务队列中
42.         timeouts.add(timeout);
43.         return timeout;
44.     }
45.    
46.     //Starts the background thread explicitly.  
47.     //The background thread will start automatically on demand even if you did not call this method.
48.     //@throws IllegalStateException if this timer has been #stop() stopped already
49.     public void start() {
50.         switch (WORKER_STATE_UPDATER.get(this)) {
51.         case WORKER_STATE_INIT:
52.             if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
53.                 //启动工作线程，即启动时间轮
54.                 workerThread.start();
55.             }
56.             break;
57.         case WORKER_STATE_STARTED:
58.             break;
59.         case WORKER_STATE_SHUTDOWN:
60.             throw new IllegalStateException("cannot be started once stopped");
61.         default:
62.             throw new Error("Invalid WorkerState");
63.         }
65.         //Wait until the startTime is initialized by the worker.
66.         while (startTime == 0) {
67.             try {
68.                 //阻塞时间轮的工作线程
69.                 startTimeInitialized.await();
70.             } catch (InterruptedException ignore) {
71.                 //Ignore - it will be ready very soon.
72.             }
73.         }
74.     }
75.     ...
76. }

复制代码

(2)实行延时使命


步调一：记载时间轮启动的时间startTime

步调二：开始do while循环，唤醒被阻塞的start()方法，通知时间轮已经启动完毕

步调三：阻塞等待下一次指针转动的时间

步调四：计算当前指针指向的时间轮的槽位idx

步调五：将已经取消的使命从HashedWheelBucket数组中移除，并将pendingTimeouts使命数 - 1

步调六：获取当前指针指向的时间槽HashedWheelBucket

步调七：遍历延时使命队列timeouts，将此中的延时使命保存到对应的槽的链表中，根据延时时间计算对应的时间槽和remainingRounds圈数

步调八：运行现在指针指向的时间槽中的链表的使命，通过taskExecutor线程池去实行到期的使命

步调九：到期的和取消的延时使命从链表中移除并将pendingTimeouts--

步调十：时间轮指针的总转动次数tick++，继续do while循环

步调十一：扫除时间轮中不需要处理惩罚的使命，保存到unprocessedTimeouts中

步调十二：将延时使命队列中还未添加到时间轮的延时使命保存到unprocessedTimeouts中

步调十三：将已经取消的使命从HashedWheelBucket数组中移除，并将pendingTimeouts使命数 - 1

1. public class HashedWheelTimer implements Timer {
2.     private volatile long startTime;
3.     private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
4.    
5.     ...
6.     //指针转动和执行延时任务的线程
7.     private final class Worker implements Runnable {
8.         //用于记录未执行的延时任务
9.         private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
10.    
11.         //总的tick数(指针嘀嗒的次数)
12.         private long tick;
13.    
14.         @Override
15.         public void run() {
16.             //1.记录时间轮启动的时间startTime
17.             startTime = System.nanoTime();
18.             if (startTime == 0) {
19.                 //我们在这里使用0作为未初始化值的指示符，所以要确保初始化时它不是0
20.                 startTime = 1;
21.             }
22.    
23.             //2.唤醒被阻塞的start()方法，通知时间轮已经启动完毕
24.             startTimeInitialized.countDown();
25.         
26.             //一直执行do while循环，直到时间轮被关闭
27.             do {
28.                 //3.阻塞等待下一次指针转动的时间
29.                 //这里会休眠tick的时间，模拟指针走动
30.                 final long deadline = waitForNextTick();
31.                 if (deadline > 0) {
32.                     //4.计算当前指针指向的时间轮槽位idx
33.                     int idx = (int) (tick & mask);
34.                     //5.将已经取消的任务从HashedWheelBucket数组中移除，并将pendingTimeouts任务数 - 1
35.                     processCancelledTasks();
36.                     //6.获取当前指针指向的时间槽HashedWheelBucket
37.                     HashedWheelBucket bucket = wheel[idx];
38.                     //7.遍历延时任务队列timeouts，将其中的延时任务保存到对应的槽的链表中
39.                     transferTimeoutsToBuckets();
40.                     //8.运行目前指针指向的槽中的链表的任务，交给taskExecutor线程池去执行到期的延时任务
41.                     //9.到期的和取消的延时任务从链表中移除并将pendingTimeouts--
42.                     bucket.expireTimeouts(deadline);
43.                     //10.时间轮指针的总转动次数tick++
44.                     tick++;
45.                 }
46.             } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
47.    
48.             //Fill the unprocessedTimeouts so we can return them from stop() method.
49.             //11.清除时间轮中不需要处理的任务
50.             for (HashedWheelBucket bucket: wheel) {
51.                 bucket.clearTimeouts(unprocessedTimeouts);
52.             }
53.             //12.将延时任务队列中还未添加到时间轮的延时任务保存到unprocessedTimeouts中
54.             //遍历任务队列，如果发现有任务被取消，则添加到unprocessedTimeouts，也就是不需要处理的队列中
55.             for (;;) {
56.                 HashedWheelTimeout timeout = timeouts.poll();
57.                 if (timeout == null) {
58.                     break;
59.                 }
60.                 if (!timeout.isCancelled()) {
61.                     //如果延时任务没被取消，记录到未执行的任务Set集合中
62.                     unprocessedTimeouts.add(timeout);
63.                 }
64.             }
65.             //13.处理被取消的任务
66.             processCancelledTasks();
67.         }
68.    
69.         //将延时任务队列timeouts中等待添加到时间轮中的延时任务，转移到时间轮的指定位置
70.         //也就是遍历延时任务队列timeouts，将其中的延时任务保存到对应的槽的链表中
71.         private void transferTimeoutsToBuckets() {
72.             //每次转移10w个延时任务
73.             for (int i = 0; i < 100000; i++) {
74.                 //从队列中出队一个延时任务
75.                 HashedWheelTimeout timeout = timeouts.poll();
76.                 if (timeout == null) {
77.                     //all processed
78.                     break;
79.                 }
80.                 if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
81.                     //Was cancelled in the meantime.
82.                     continue;
83.                 }
84.                 //到期一共需要走多少时间格(tick次数)，deadline表示当前任务的延迟时间(从时间轮启动时计算)，tickDuration表示时间格的时间间隔
85.                 long calculated = timeout.deadline / tickDuration;
86.                 //tick已经走了的时间格，到期一共还需要需要走多少圈
87.                 timeout.remainingRounds = (calculated - tick) / wheel.length;
88.                 //如果延时任务在队列中等待太久已经过了执行时间，那么这个时候就使用当前tick，也就是放在当前的bucket，此方法调用完后就会被执行
89.                 final long ticks = Math.max(calculated, tick);
90.                 //槽的索引，stopIndex = tick 次数 & mask, mask = wheel.length - 1
91.                 int stopIndex = (int) (ticks & mask);
92.                 //根据索引该任务应该放到的槽
93.                 HashedWheelBucket bucket = wheel[stopIndex];
94.                 //将任务添加到槽中，链表末尾
95.                 bucket.addTimeout(timeout);
96.             }
97.         }
98.    
99.         //处理取消掉的延时任务
100.         //将已经取消的任务从HashedWheelBucket数组中移除，并将pendingTimeouts任务数 - 1
101.         private void processCancelledTasks() {
102.             for (;;) {
103.                 HashedWheelTimeout timeout = cancelledTimeouts.poll();
104.                 if (timeout == null) {
105.                     //all processed
106.                     break;
107.                 }
108.                 try {
109.                     timeout.remove();
110.                 } catch (Throwable t) {
111.                     if (logger.isWarnEnabled()) {
112.                         logger.warn("An exception was thrown while process a cancellation task", t);
113.                     }
114.                 }
115.             }
116.         }
117.    
118.         //从时间轮的启动时间startTime和当前的tick数(指针跳动次数)计算下一次指针跳动的时间，然后休眠等待下一次指针跳动时间到来
119.         private long waitForNextTick() {
120.             //deadline返回的是下一次时间轮指针跳动的时间与时间格启动的时间间隔
121.             long deadline = tickDuration * (tick + 1);
122.    
123.             for (;;) {
124.                 //计算当前时间距离启动时间的时间间隔
125.                 final long currentTime = System.nanoTime() - startTime;
126.                 //距离下一次指针跳动还需休眠多长时间
127.                 long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
128.                 //到了指针调到下一个槽位的时间
129.                 if (sleepTimeMs <= 0) {
130.                     if (currentTime == Long.MIN_VALUE) {
131.                         return -Long.MAX_VALUE;
132.                     } else {
133.                         return currentTime;
134.                     }
135.                 }
136.    
137.                 try {
138.                     //表示距离下一次指针跳动还需要一段时间，所以休眠等待时间的到来
139.                     Thread.sleep(sleepTimeMs);
140.                 } catch (InterruptedException ignored) {
141.                     if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
142.                         return Long.MIN_VALUE;
143.                     }
144.                 }
145.             }
146.         }
147.    
148.         //记录未执行的延时任务
149.         public Set<Timeout> unprocessedTimeouts() {
150.             return Collections.unmodifiableSet(unprocessedTimeouts);
151.         }
152.     }
153.    
154.     private static final class HashedWheelBucket {
155.         ...
156.         public void expireTimeouts(long deadline) {
157.             HashedWheelTimeout timeout = head;
158.         
159.             //process all timeouts
160.             while (timeout != null) {
161.                 HashedWheelTimeout next = timeout.next;
162.                 if (timeout.remainingRounds <= 0) {
163.                     next = remove(timeout);
164.                     if (timeout.deadline <= deadline) {
165.                         //通过线程池执行任务
166.                         timeout.expire();
167.                     } else {
168.                         //The timeout was placed into a wrong slot. This should never happen.
169.                         throw new IllegalStateException(String.format("timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
170.                     }
171.                 } else if (timeout.isCancelled()) {
172.                     next = remove(timeout);
173.                 } else {
174.                     timeout.remainingRounds --;
175.                 }
176.                 timeout = next;
177.             }
178.         }
180.         public HashedWheelTimeout remove(HashedWheelTimeout timeout) {
181.             HashedWheelTimeout next = timeout.next;
182.             //remove timeout that was either processed or cancelled by updating the linked-list
183.             if (timeout.prev != null) {
184.                 timeout.prev.next = next;
185.             }
186.             if (timeout.next != null) {
187.                 timeout.next.prev = timeout.prev;
188.             }
190.             if (timeout == head) {
191.                 //if timeout is also the tail we need to adjust the entry too
192.                 if (timeout == tail) {
193.                     tail = null;
194.                     head = null;
195.                 } else {
196.                     head = next;
197.                 }
198.             } else if (timeout == tail) {
199.                 //if the timeout is the tail modify the tail to be the prev node.
200.                 tail = timeout.prev;
201.             }
202.             //null out prev, next and bucket to allow for GC.
203.             timeout.prev = null;
204.             timeout.next = null;
205.             timeout.bucket = null;
206.             timeout.timer.pendingTimeouts.decrementAndGet();
207.             return next;
208.         }
209.         ...   
210.     }
211.    
212.     private static final class HashedWheelTimeout implements Timeout, Runnable {
213.         private final TimerTask task;
214.         private final HashedWheelTimer timer;
215.         ...
216.         public void expire() {
217.             if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {
218.                 return;
219.             }
220.             try {
221.                 timer.taskExecutor.execute(this);
222.             } catch (Throwable t) {
223.                 if (logger.isWarnEnabled()) {
224.                     logger.warn("An exception was thrown while submit " + TimerTask.class.getSimpleName() + " for execution.", t);
225.                 }
226.             }
227.         }
228.         ...
229.     }
230.     ...
231. }

复制代码

8.HashedWheelTimer的完整源码

1. //Netty时间轮
2. public class HashedWheelTimer implements Timer {
3.     static final InternalLogger logger = InternalLoggerFactory.getInstance(HashedWheelTimer.class);
4.     private static final AtomicInteger INSTANCE_COUNTER = new AtomicInteger();
5.     private static final AtomicBoolean WARNED_TOO_MANY_INSTANCES = new AtomicBoolean();
6.     private static final int INSTANCE_COUNT_LIMIT = 64;
7.     private static final long MILLISECOND_NANOS = TimeUnit.MILLISECONDS.toNanos(1);
8.     private static final ResourceLeakDetector<HashedWheelTimer> leakDetector =
9.         ResourceLeakDetectorFactory.instance().newResourceLeakDetector(HashedWheelTimer.class, 1);
10.     private static final AtomicIntegerFieldUpdater<HashedWheelTimer> WORKER_STATE_UPDATER =
11.         AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimer.class, "workerState");
12.     private final ResourceLeakTracker<HashedWheelTimer> leak;
13.     //指针转动和延时任务执行的线程
14.     private final Worker worker = new Worker();
15.     //worker任务封装的工作线程，用于指针转动和触发时间格里的延时任务的执行
16.     private final Thread workerThread;
18.     public static final int WORKER_STATE_INIT = 0;
19.     public static final int WORKER_STATE_STARTED = 1;
20.     public static final int WORKER_STATE_SHUTDOWN = 2;
21.     @SuppressWarnings({"unused", "FieldMayBeFinal"})
22.     private volatile int workerState;//0 - init, 1 - started, 2 - shut down
23.     //每个时间格的时间跨度，默认为100ms
24.     private final long tickDuration;
25.     //时间轮(环形数组)，HashedWheelBucket为每个时间格的槽
26.     private final HashedWheelBucket[] wheel;
27.     private final int mask;
28.     private final CountDownLatch startTimeInitialized = new CountDownLatch(1);
29.     //延时任务队列，队列中为等待被添加到时间轮的延时任务
30.     private final Queue<HashedWheelTimeout> timeouts = PlatformDependent.newMpscQueue();
31.     //保存已经取消的延时任务的队列
32.     private final Queue<HashedWheelTimeout> cancelledTimeouts = PlatformDependent.newMpscQueue();
33.     //记录当前的任务数
34.     private final AtomicLong pendingTimeouts = new AtomicLong(0);
35.     //最大的任务数
36.     private final long maxPendingTimeouts;
37.     //执行延时任务的线程池
38.     private final Executor taskExecutor;
39.     //工作线程启动时间
40.     private volatile long startTime;
42.     // 构造器 start //
43.     public HashedWheelTimer() {
44.         this(Executors.defaultThreadFactory());
45.     }
47.     public HashedWheelTimer(long tickDuration, TimeUnit unit) {
48.         this(Executors.defaultThreadFactory(), tickDuration, unit);
49.     }
51.     public HashedWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) {
52.         this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel);
53.     }
55.     //使用默认的tickDuration(时间格跨度默认为100ms)和默认的ticksPerWheel(时间格总数默认为512)创建一个新的计时器(时间轮)
56.     public HashedWheelTimer(ThreadFactory threadFactory) {
57.         this(threadFactory, 100, TimeUnit.MILLISECONDS);
58.     }
60.     public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit) {
61.         this(threadFactory, tickDuration, unit, 512);
62.     }
64.     public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel) {
65.         this(threadFactory, tickDuration, unit, ticksPerWheel, true);
66.     }
68.     public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection) {
69.         this(threadFactory, tickDuration, unit, ticksPerWheel, leakDetection, -1);
70.     }
72.     public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection, long maxPendingTimeouts) {
73.         this(threadFactory, tickDuration, unit, ticksPerWheel, leakDetection, maxPendingTimeouts, ImmediateExecutor.INSTANCE);
74.     }
76.     //Creates a new timer.
77.     //@param threadFactory        创建线程的工厂
78.     //@param tickDuration         每格的时间间隔，默认100ms，0.1秒
79.     //@param unit                 时间单位，默认为毫秒
80.     //@param ticksPerWheel        时间轮的格子数，默认为512；如果传入的不是2的N次方，则会调整为大于等于该参数的第一个2的N次方，好处是可以优化hash值的计算
81.     //@param leakDetection        如果false，那么只有工作线程不是后台线程时才会追踪资源泄露，这个参数可以忽略
82.     //@param maxPendingTimeouts   最大的pending数量（时间轮中任务的最大数量），超过这个值之后调用将抛出异常，0或者负数表示没有限制，默认为-1
83.     //@param taskExecutor         任务线程池，用于执行提交的任务，调用者负责在不需要时关闭它
84.     //@throws NullPointerException     if either of threadFactory and unit is null
85.     //@throws IllegalArgumentException if either of tickDuration and ticksPerWheel is <= 0
86.     public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection, long maxPendingTimeouts, Executor taskExecutor) {
87.         //1.构造参数校验及给实际执行延时任务的线程池taskExecutor赋值
88.         checkNotNull(threadFactory, "threadFactory");
89.         checkNotNull(unit, "unit");
90.         checkPositive(tickDuration, "tickDuration");
91.         checkPositive(ticksPerWheel, "ticksPerWheel");
92.         this.taskExecutor = checkNotNull(taskExecutor, "taskExecutor");
94.         //2.将ticksPerWheel(时间轮上的时间格数)向上取值为2的幂，方便进行求商和取余计算
95.         //3.初始化时间轮wheel
96.         wheel = createWheel(ticksPerWheel);
97.         //mask的设计和HashMap一样，通过限制数组的大小为2的幂，利用位运算来替代取模运算，提高性能
98.         mask = wheel.length - 1;
99.       
100.         //4.校验tickDuration和ticksPerWheel
101.         //Convert tickDuration to nanos.
102.         long duration = unit.toNanos(tickDuration);
103.         //防止溢出
104.         //tickDuration * ticksPerWheel必须小于Long.MAX_VALUE
105.         if (duration >= Long.MAX_VALUE / wheel.length) {
106.             throw new IllegalArgumentException(String.format("tickDuration: %d (expected: 0 < tickDuration in nanos < %d", tickDuration, Long.MAX_VALUE / wheel.length));
107.         }
108.         //tickDuration不能小于1ms
109.         if (duration < MILLISECOND_NANOS) {
110.             logger.warn("Configured tickDuration {} smaller then {}, using 1ms.", tickDuration, MILLISECOND_NANOS);
111.             this.tickDuration = MILLISECOND_NANOS;
112.         } else {
113.             this.tickDuration = duration;
114.         }
115.         //5.创建工作线程，用于指针转动和触发时间格里的延时任务的执行
116.         workerThread = threadFactory.newThread(worker);
118.         leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;
119.         //6.给时间轮中任务的最大数量maxPendingTimeouts赋值
120.         this.maxPendingTimeouts = maxPendingTimeouts;
121.         //7.检查HashedWheelTimer的实例数量，如果大于64则打印error日志
122.         if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
123.             WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
124.             reportTooManyInstances();
125.         }
126.     }
127.     // 构造器 end //
128.    
129.     @Override
130.     protected void finalize() throws Throwable {
131.         try {
132.             super.finalize();
133.         } finally {
134.             //This object is going to be GCed and it is assumed the ship has sailed to do a proper shutdown.
135.             //If we have not yet shutdown then we want to make sure we decrement the active instance count.
136.             if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
137.                 INSTANCE_COUNTER.decrementAndGet();
138.             }
139.         }
140.     }
142.     //初始化时间轮环形数组
143.     //@param ticksPerWheel
144.     private static HashedWheelBucket[] createWheel(int ticksPerWheel) {
145.         //ticksPerWheel不能大于2^30
146.         checkInRange(ticksPerWheel, 1, 1073741824, "ticksPerWheel");
147.         //将ticksPerWheel（轮子上的时间格数）向上取值为2的次幂
148.         ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
149.         //创建时间轮环形数组
150.         HashedWheelBucket[] wheel = new HashedWheelBucket[ticksPerWheel];
151.         for (int i = 0; i < wheel.length; i ++) {
152.             wheel[i] = new HashedWheelBucket();
153.         }
154.         return wheel;
155.     }
157.     //将ticksPerWheel(时间轮上的时间格数)向上取值为2的次幂
158.     private static int normalizeTicksPerWheel(int ticksPerWheel) {
159.         int normalizedTicksPerWheel = 1;
160.         while (normalizedTicksPerWheel < ticksPerWheel) {
161.             normalizedTicksPerWheel <<= 1;
162.         }
163.         return normalizedTicksPerWheel;
164.     }
166.     //显式启动后台线程
167.     //即使没有调用此方法，后台线程也会按需自动启动
168.     //Starts the background thread explicitly.  
169.     //The background thread will start automatically on demand even if you did not call this method.
170.     //@throws IllegalStateException if this timer has been #stop() stopped already
171.     public void start() {
172.         switch (WORKER_STATE_UPDATER.get(this)) {
173.         case WORKER_STATE_INIT:
174.             if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
175.                 //启动工作线程，即启动时间轮
176.                 workerThread.start();
177.             }
178.             break;
179.         case WORKER_STATE_STARTED:
180.             break;
181.         case WORKER_STATE_SHUTDOWN:
182.             throw new IllegalStateException("cannot be started once stopped");
183.         default:
184.             throw new Error("Invalid WorkerState");
185.         }
187.         //Wait until the startTime is initialized by the worker.
188.         while (startTime == 0) {
189.             try {
190.                 //阻塞时间轮的工作线程
191.                 startTimeInitialized.await();
192.             } catch (InterruptedException ignore) {
193.                 //Ignore - it will be ready very soon.
194.             }
195.         }
196.     }
198.     @Override
199.     public Set<Timeout> stop() {
200.         if (Thread.currentThread() == workerThread) {
201.             throw new IllegalStateException(HashedWheelTimer.class.getSimpleName() + ".stop() cannot be called from " + TimerTask.class.getSimpleName());
202.         }
204.         if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
205.             //workerState can be 0 or 2 at this moment - let it always be 2.
206.             if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
207.                 INSTANCE_COUNTER.decrementAndGet();
208.                 if (leak != null) {
209.                     boolean closed = leak.close(this);
210.                     assert closed;
211.                 }
212.             }
213.             return Collections.emptySet();
214.         }
216.         try {
217.             boolean interrupted = false;
218.             while (workerThread.isAlive()) {
219.                 workerThread.interrupt();
220.                 try {
221.                     workerThread.join(100);
222.                 } catch (InterruptedException ignored) {
223.                     interrupted = true;
224.                 }
225.             }
226.             if (interrupted) {
227.                 Thread.currentThread().interrupt();
228.             }
229.         } finally {
230.             INSTANCE_COUNTER.decrementAndGet();
231.             if (leak != null) {
232.                 boolean closed = leak.close(this);
233.                 assert closed;
234.             }
235.         }
236.         return worker.unprocessedTimeouts();
237.     }
239.     //添加延时任务
240.     //@param task 任务
241.     //@param delay 延时时间
242.     //@param unit 延时时间单位
243.     @Override
244.     public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
245.         checkNotNull(task, "task");
246.         checkNotNull(unit, "unit");
247.         //1.将需要执行的延时任务数pendingTimeouts + 1
248.         long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();
249.       
250.         //2.如果pendingTimeouts超过maxPendingTimeouts，则抛出异常
251.         if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
252.             pendingTimeouts.decrementAndGet();
253.             throw new RejectedExecutionException("Number of pending timeouts (" + pendingTimeoutsCount
254.                 + ") is greater than or equal to maximum allowed pending " + "timeouts (" + maxPendingTimeouts + ")");
255.         }
256.       
257.         //3.启动工作线程，即启动时间轮
258.         start();
259.      
260.         //将延时任务添加到延时任务队列timeouts中，该队列将在下一个滴答声中处理(指针的下一次转动)
261.         //在处理过程中，所有排队的HashedWheelTimeout将被添加到正确的HashedWheelBucket
263.         //4.计算任务的截止时间deadline = 当前时间 + 当前任务执行的延迟时间 - 时间轮启动的时间
264.         long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;
265.         if (delay > 0 && deadline < 0) {
266.             deadline = Long.MAX_VALUE;
267.         }
268.       
269.         //5.创建延时任务实例HashedWheelTimeout
270.         HashedWheelTimeout timeout = new HashedWheelTimeout(this, task, deadline);
271.       
272.         //6.将延时任务实例添加到延时任务队列中
273.         timeouts.add(timeout);
274.         return timeout;
275.     }
277.     //Returns the number of pending timeouts of this Timer.
278.     public long pendingTimeouts() {
279.         return pendingTimeouts.get();
280.     }
282.     private static void reportTooManyInstances() {
283.         if (logger.isErrorEnabled()) {
284.             String resourceType = simpleClassName(HashedWheelTimer.class);
285.             logger.error("You are creating too many " + resourceType + " instances. " +
286.                 resourceType + " is a shared resource that must be reused across the JVM, " +
287.                 "so that only a few instances are created.");
288.         }
289.     }
291.     //指针转动和延时任务执行的线程
292.     private final class Worker implements Runnable {
293.         //用于记录未执行的延时任务
294.         private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
295.    
296.         //总的tick数(指针嘀嗒的次数)
297.         private long tick;
298.    
299.         @Override
300.         public void run() {
301.             //1.记录时间轮启动的时间startTime
302.             startTime = System.nanoTime();
303.             if (startTime == 0) {
304.                 //我们在这里使用0作为未初始化值的指示符，所以要确保初始化时它不是0
305.                 startTime = 1;
306.             }
307.    
308.             //2.唤醒被阻塞的start()方法，通知时间轮已经启动完毕
309.             startTimeInitialized.countDown();
310.         
311.             //一直执行do while循环，直到时间轮被关闭
312.             do {
313.                 //3.阻塞等待下一次指针转动的时间
314.                 //这里会休眠tick的时间，模拟指针走动
315.                 final long deadline = waitForNextTick();
316.                 if (deadline > 0) {
317.                     //4.计算当前指针指向的时间轮槽位idx
318.                     int idx = (int) (tick & mask);
319.                     //5.将已经取消的任务从HashedWheelBucket数组中移除，并将pendingTimeouts任务数 - 1
320.                     processCancelledTasks();
321.                     //6.获取当前指针指向的时间槽HashedWheelBucket
322.                     HashedWheelBucket bucket = wheel[idx];
323.                     //7.遍历延时任务队列timeouts，将其中的延时任务保存到对应的槽的链表中
324.                     transferTimeoutsToBuckets();
325.                     //8.运行目前指针指向的槽中的链表的任务，交给taskExecutor线程池去执行到期的延时任务
326.                     //9.到期的和取消的延时任务从链表中移除并将pendingTimeouts--
327.                     bucket.expireTimeouts(deadline);
328.                     //10.时间轮指针的总转动次数tick++
329.                     tick++;
330.                 }
331.             } while (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_STARTED);
332.    
333.             //Fill the unprocessedTimeouts so we can return them from stop() method.
334.             //11.清除时间轮中不需要处理的任务
335.             for (HashedWheelBucket bucket: wheel) {
336.                 bucket.clearTimeouts(unprocessedTimeouts);
337.             }
338.             //12.将延时任务队列中还未添加到时间轮的延时任务保存到unprocessedTimeouts中
339.             //遍历任务队列，如果发现有任务被取消，则添加到unprocessedTimeouts，也就是不需要处理的队列中
340.             for (;;) {
341.                 HashedWheelTimeout timeout = timeouts.poll();
342.                 if (timeout == null) {
343.                     break;
344.                 }
345.                 if (!timeout.isCancelled()) {
346.                     //如果延时任务没被取消，记录到未执行的任务Set集合中
347.                     unprocessedTimeouts.add(timeout);
348.                 }
349.             }
350.             //13.处理被取消的任务
351.             processCancelledTasks();
352.         }
353.    
354.         //将延时任务队列timeouts中等待添加到时间轮中的延时任务，转移到时间轮的指定位置
355.         //也就是遍历延时任务队列timeouts，将其中的延时任务保存到对应的槽的链表中
356.         private void transferTimeoutsToBuckets() {
357.             //每次转移10w个延时任务
358.             for (int i = 0; i < 100000; i++) {
359.                 //从队列中出队一个延时任务
360.                 HashedWheelTimeout timeout = timeouts.poll();
361.                 if (timeout == null) {
362.                     //all processed
363.                     break;
364.                 }
365.                 if (timeout.state() == HashedWheelTimeout.ST_CANCELLED) {
366.                     //Was cancelled in the meantime.
367.                     continue;
368.                 }
369.                 //到期一共需要走多少时间格(tick次数)，deadline表示当前任务的延迟时间(从时间轮启动时计算)，tickDuration表示时间格的时间间隔
370.                 long calculated = timeout.deadline / tickDuration;
371.                 //tick已经走了的时间格，到期一共还需要需要走多少圈
372.                 timeout.remainingRounds = (calculated - tick) / wheel.length;
373.                 //如果延时任务在队列中等待太久已经过了执行时间，那么这个时候就使用当前tick，也就是放在当前的bucket，此方法调用完后就会被执行
374.                 final long ticks = Math.max(calculated, tick);
375.                 //槽的索引，stopIndex = tick 次数 & mask, mask = wheel.length - 1
376.                 int stopIndex = (int) (ticks & mask);
377.                 //根据索引该任务应该放到的槽
378.                 HashedWheelBucket bucket = wheel[stopIndex];
379.                 //将任务添加到槽中，链表末尾
380.                 bucket.addTimeout(timeout);
381.             }
382.         }
383.    
384.         //处理取消掉的延时任务
385.         //将已经取消的任务从HashedWheelBucket数组中移除，并将pendingTimeouts任务数 - 1
386.         private void processCancelledTasks() {
387.             for (;;) {
388.                 HashedWheelTimeout timeout = cancelledTimeouts.poll();
389.                 if (timeout == null) {
390.                     //all processed
391.                     break;
392.                 }
393.                 try {
394.                     timeout.remove();
395.                 } catch (Throwable t) {
396.                     if (logger.isWarnEnabled()) {
397.                         logger.warn("An exception was thrown while process a cancellation task", t);
398.                     }
399.                 }
400.             }
401.         }
402.    
403.         //从时间轮的启动时间startTime和当前的tick数(指针跳动次数)计算下一次指针跳动的时间，然后休眠等待下一次指针跳动时间到来
404.         private long waitForNextTick() {
405.             //deadline返回的是下一次时间轮指针跳动的时间与时间格启动的时间间隔
406.             long deadline = tickDuration * (tick + 1);
407.    
408.             for (;;) {
409.                 //计算当前时间距离启动时间的时间间隔
410.                 final long currentTime = System.nanoTime() - startTime;
411.                 //距离下一次指针跳动还需休眠多长时间
412.                 long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;
413.                 //到了指针调到下一个槽位的时间
414.                 if (sleepTimeMs <= 0) {
415.                     if (currentTime == Long.MIN_VALUE) {
416.                         return -Long.MAX_VALUE;
417.                     } else {
418.                         return currentTime;
419.                     }
420.                 }
421.    
422.                 try {
423.                     //表示距离下一次指针跳动还需要一段时间，所以休眠等待时间的到来
424.                     Thread.sleep(sleepTimeMs);
425.                 } catch (InterruptedException ignored) {
426.                     if (WORKER_STATE_UPDATER.get(HashedWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
427.                         return Long.MIN_VALUE;
428.                     }
429.                 }
430.             }
431.         }
432.    
433.         //记录未执行的延时任务
434.         public Set<Timeout> unprocessedTimeouts() {
435.             return Collections.unmodifiableSet(unprocessedTimeouts);
436.         }
437.     }
438.    
439.     //延时任务
440.     private static final class HashedWheelTimeout implements Timeout, Runnable {
441.         private static final int ST_INIT = 0;
442.         private static final int ST_CANCELLED = 1;
443.         private static final int ST_EXPIRED = 2;
444.         private static final AtomicIntegerFieldUpdater<HashedWheelTimeout> STATE_UPDATER =
445.             AtomicIntegerFieldUpdater.newUpdater(HashedWheelTimeout.class, "state");
447.         private final HashedWheelTimer timer;
448.         private final TimerTask task;
449.         //任务执行的截止时间 = 当前时间 + 延时任务延时时间 - 时间轮启动时间
450.         private final long deadline;
452.         @SuppressWarnings({"unused", "FieldMayBeFinal", "RedundantFieldInitialization" })
453.         private volatile int state = ST_INIT;
455.         //剩下的圈(轮)数
456.         //remainingRounds将由Worker.transferTimeoutsToBuckets()在HashedWheelTimeout被添加到正确的HashedWheelBucket之前计算和设置
457.         long remainingRounds;
459.         //HashedWheelTimerBucket槽中的延时任务列表是一个双向链表
460.         //因为只有workerThread会对它进行操作，所以不需要 synchronization / volatile
461.         HashedWheelTimeout next;
462.         HashedWheelTimeout prev;
464.         //当前延时任务所插入时间轮的哪个槽
465.         HashedWheelBucket bucket;
467.         HashedWheelTimeout(HashedWheelTimer timer, TimerTask task, long deadline) {
468.             this.timer = timer;
469.             this.task = task;
470.             this.deadline = deadline;
471.         }
473.         @Override
474.         public Timer timer() {
475.             return timer;
476.         }
478.         @Override
479.         public TimerTask task() {
480.             return task;
481.         }
483.         @Override
484.         public boolean cancel() {
485.             //only update the state it will be removed from HashedWheelBucket on next tick.
486.             if (!compareAndSetState(ST_INIT, ST_CANCELLED)) {
487.                 return false;
488.             }
489.             //If a task should be canceled we put this to another queue which will be processed on each tick.
490.             //So this means that we will have a GC latency of max. 1 tick duration which is good enough. This way
491.             //we can make again use of our MpscLinkedQueue and so minimize the locking / overhead as much as possible.
492.             timer.cancelledTimeouts.add(this);
493.             return true;
494.         }
496.         void remove() {
497.             HashedWheelBucket bucket = this.bucket;
498.             if (bucket != null) {
499.                 bucket.remove(this);
500.             } else {
501.                 timer.pendingTimeouts.decrementAndGet();
502.             }
503.         }
505.         public boolean compareAndSetState(int expected, int state) {
506.             return STATE_UPDATER.compareAndSet(this, expected, state);
507.         }
509.         public int state() {
510.             return state;
511.         }
513.         @Override
514.         public boolean isCancelled() {
515.             return state() == ST_CANCELLED;
516.         }
518.         @Override
519.         public boolean isExpired() {
520.             return state() == ST_EXPIRED;
521.         }
523.         public void expire() {
524.             if (!compareAndSetState(ST_INIT, ST_EXPIRED)) {
525.                 return;
526.             }
528.             try {
529.                 timer.taskExecutor.execute(this);
530.             } catch (Throwable t) {
531.                 if (logger.isWarnEnabled()) {
532.                     logger.warn("An exception was thrown while submit " + TimerTask.class.getSimpleName() + " for execution.", t);
533.                 }
534.             }
535.         }
537.         @Override
538.         public void run() {
539.             try {
540.                 task.run(this);
541.             } catch (Throwable t) {
542.                 if (logger.isWarnEnabled()) {
543.                     logger.warn("An exception was thrown by " + TimerTask.class.getSimpleName() + '.', t);
544.                 }
545.             }
546.         }
548.         @Override
549.         public String toString() {
550.             final long currentTime = System.nanoTime();
551.             long remaining = deadline - currentTime + timer.startTime;
553.             StringBuilder buf = new StringBuilder(192).append(simpleClassName(this)).append('(').append("deadline: ");
554.             if (remaining > 0) {
555.                 buf.append(remaining).append(" ns later");
556.             } else if (remaining < 0) {
557.                 buf.append(-remaining).append(" ns ago");
558.             } else {
559.                 buf.append("now");
560.             }
561.             if (isCancelled()) {
562.                 buf.append(", cancelled");
563.             }
564.             return buf.append(", task: ").append(task()).append(')').toString();
565.         }
566.     }
568.     //存放HashedWheelTimeouts的桶
569.     //这些数据存储在一个类似于链表的数据结构中，允许轻松删除中间的hashedwheeltimeout
570.     //HashedWheelTimeout本身作为节点，因此不需要创建额外的对象
571.     //保存头结点和尾节点，方便于任务的提取和插入
572.     private static final class HashedWheelBucket {
573.         //头结点
574.         private HashedWheelTimeout head;
575.         //尾节点
576.         private HashedWheelTimeout tail;
578.         //Add HashedWheelTimeout to this bucket.
579.         public void addTimeout(HashedWheelTimeout timeout) {
580.             assert timeout.bucket == null;
581.             timeout.bucket = this;
582.             if (head == null) {
583.                 head = tail = timeout;
584.             } else {
585.                 tail.next = timeout;
586.                 timeout.prev = tail;
587.                 tail = timeout;
588.             }
589.         }
591.         //Expire all HashedWheelTimeouts for the given deadline.
592.         public void expireTimeouts(long deadline) {
593.             HashedWheelTimeout timeout = head;
594.             //遍历当前时间槽中的所有任务
595.             while (timeout != null) {
596.                 HashedWheelTimeout next = timeout.next;
597.                 if (timeout.remainingRounds <= 0) {
598.                     //从链表中移除
599.                     next = remove(timeout);
600.                     if (timeout.deadline <= deadline) {
601.                         //延时任务到期,执行延时任务
602.                         timeout.expire();
603.                     } else {
604.                         //The timeout was placed into a wrong slot. This should never happen.
605.                         throw new IllegalStateException(String.format("timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
606.                     }
607.                     //如果延时任务取消，从链表中移除
608.                 } else if (timeout.isCancelled()) {
609.                     next = remove(timeout);
610.                 } else {
611.                     //任务还没到期，剩余的轮数-1
612.                     timeout.remainingRounds --;
613.                 }
614.                 //将指针放置到下一个延时任务上
615.                 timeout = next;
616.             }
617.         }
619.         //删除槽中链表中的延时任务
620.         public HashedWheelTimeout remove(HashedWheelTimeout timeout) {
621.             HashedWheelTimeout next = timeout.next;
622.             //remove timeout that was either processed or cancelled by updating the linked-list
623.             if (timeout.prev != null) {
624.                 timeout.prev.next = next;
625.             }
626.             if (timeout.next != null) {
627.                 timeout.next.prev = timeout.prev;
628.             }
629.             if (timeout == head) {
630.                 //if timeout is also the tail we need to adjust the entry too
631.                 if (timeout == tail) {
632.                     tail = null;
633.                     head = null;
634.                 } else {
635.                     head = next;
636.                 }
637.             } else if (timeout == tail) {
638.                 //if the timeout is the tail modify the tail to be the prev node.
639.                 tail = timeout.prev;
640.             }
641.             //null out prev, next and bucket to allow for GC.
642.             timeout.prev = null;
643.             timeout.next = null;
644.             timeout.bucket = null;
645.             timeout.timer.pendingTimeouts.decrementAndGet();
646.             return next;
647.         }
649.         //Clear this bucket and return all not expired / cancelled Timeouts.
650.         public void clearTimeouts(Set<Timeout> set) {
651.             for (;;) {
652.                 HashedWheelTimeout timeout = pollTimeout();
653.                 if (timeout == null) {
654.                     return;
655.                 }
656.                 if (timeout.isExpired() || timeout.isCancelled()) {
657.                     continue;
658.                 }
659.                 set.add(timeout);
660.             }
661.         }
663.         //头结点移除
664.         private HashedWheelTimeout pollTimeout() {
665.             HashedWheelTimeout head = this.head;
666.             if (head == null) {
667.                 return null;
668.             }
669.             HashedWheelTimeout next = head.next;
670.             if (next == null) {
671.                 tail = this.head =  null;
672.             } else {
673.                 this.head = next;
674.                 next.prev = null;
675.             }
677.             //null out prev and next to allow for GC.
678.             head.next = null;
679.             head.prev = null;
680.             head.bucket = null;
681.             return head;
682.         }
683.     }
684. }

复制代码

9.HashedWheelTimer的总结


一.时间轮的转动是单线程
但是时间轮中每个时间槽里的延时使命则是由线程池来实行的。

二.延时使命保存到JVM中没有做宕机备份
系统重启时延时使命将会丢失，无法恢复使命进行重新调度。

三.时间轮调度器的时间精度不是很高
对于精度要求特殊高的调度使命可能不太适合，由于时间轮的精度取决于时间格的跨度大小。

四.时间轮指针的转动是使用Sleep来完成等待的

10.HashedWheelTimer的应用


(1)时间轮的应用场景


一.Dubbo、Netty、Kafka、Redission等中间件都用到了时间轮机制
二.订单关闭、确认收货、批量定时数据更新等都可以接纳时间轮机制

(2)心跳检测


心跳机制会每隔固定的时间发送一个心跳包来检测客户端和服务端的连接状态，客户端发送心跳包用来告诉服务器其还正常运行。

好比在Dubbo中，需要有心跳机制来维持Consumer与Provider的长连接，默认的心跳隔断是60s。当Provider在3次心跳时间内没有收到心跳相应，会关闭连接通道。当Consumer在3次心跳时间内没有收到心跳相应，会进行重连。

在Dubbo的HeaderExchangeClient类中会向时间轮中提交该心跳使命：

一.发送心跳的时间轮

1. private static final HashedWheelTimer IDLE_CHECK_TIMER =
2.     new HashedWheelTimer(
3.         new NamedThreadFactory("dubbo-client-idleCheck", true),
4.         1,
5.         TimeUnit.SECONDS,
6.         TICKS_PER_WHEEL
7.     );

复制代码

二.向时间轮中提交心跳使命

1. private void startHeartBeatTask(URL url) {
2.     //Client的具体实现决定是否启动该心跳任务
3.     if (!client.canHandleIdle()) {
4.         AbstractTimerTask.ChannelProvider cp = () -> Collections.singletonList(HeaderExchangeClient.this);
5.         //计算心跳间隔, 最小间隔不能低于1s
6.         int heartbeat = getHeartbeat(url);
7.         long heartbeatTick = calculateLeastDuration(heartbeat);
8.         //创建心跳任务
9.         this.heartBeatTimerTask = new HeartbeatTimerTask(cp, heartbeatTick, heartbeat);
10.         //提交到IDLE_CHECK_TIMER这个时间轮中等待执行, 等时间到了时间轮就会去取出该任务进行调度执行
11.         IDLE_CHECK_TIMER.newTimeout(heartBeatTimerTask, heartbeatTick, TimeUnit.MILLISECONDS);
12.     }
13. }

复制代码

(3)超时处理惩罚


在Dubbo中发起RPC调用时，通常会设置超时时间，当消耗者调用服务提供者出现超时进行一定的逻辑处理惩罚。那么怎么检测使命调用超时了呢？我们可以使用定时使命。

每次发起RPC调用时创建一个Future，记载这个Future的创建时间与超时时间，后台有一个定时使命进行检测。当Future到达超时时间而且没有被处理惩罚时，就需要对这个Future实行超时逻辑处理惩罚。

(4)Redisson分布式锁续期


Redisson看门狗机制，通过时间轮定时给分布式锁续期。在获取锁乐成后，Redisson会封装一个锁续期的延时使命放入到时间轮中。默认10秒查抄一下，用于对获取到的锁进行续期，延长持有锁的时间。如果业务机器宕机了，那么续期的延时使命失效，也无法续期，锁会超时开释。

一.添加续期延时使命

1. private void renewExpiration() {
2.     ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());
3.     if (ee == null) {
4.         return;
5.     }
7.     //这边newTimeout点进去发现就是往时间轮中提交了一个任务
8.     Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
9.         @Override
10.         public void run(Timeout timeout) throws Exception {
11.             ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());
12.             if (ent == null) {
13.                 return;
14.             }
15.             Long threadId = ent.getFirstThreadId();
16.             if (threadId == null) {
17.                 return;
18.             }
20.             RFuture<Boolean> future = renewExpirationAsync(threadId);
21.             future.onComplete((res, e) -> {
22.                 if (e != null) {
23.                     log.error("Can't update lock " + getName() + " expiration", e);
24.                     return;
25.                 }
26.                 if (res) {
27.                     //续期成功后继续调度, 又往时间轮中放一个续期任务
28.                     renewExpiration();
29.                 }
30.             });
31.         }
32.     }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
33.     ee.setTimeout(task);
34. }

复制代码

二.lua续期代码

1. protected RFuture<Boolean> renewExpirationAsync(long threadId) {
2.     //通过lua脚本对锁进行续期
3.     return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
4.         "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
5.         "redis.call('pexpire', KEYS[1], ARGV[1]); " +
6.         "return 1; " +
7.         "end; " +
8.         "return 0;",
9.         Collections.singletonList(getName()),
10.         internalLockLeaseTime, getLockName(threadId)
11.     );
12. }

复制代码

   文章转载自：东阳马生架构
  原文链接：Netty源码—10.Netty工具之时间轮 - 东阳马生架构 - 博客园
  体验地点：引迈 - JNPF快速开发平台_低代码开发平台_零代码开发平台_流程计划器_表单引擎_工作流引擎_软件架构

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。