时间轮使用场景（时间轮是什么）

上一篇我们讲了定时器的几种实现，分析了在大数据量高并发的场景下这几种实现方式就有点力不从心了，从而引出时间轮这种数据结构。在netty 和kafka 这两种优秀的中间件中，都有时间轮的实现。文章最后，我们模拟kafka 中scala 的代码实现java版的时间轮。

Netty 的时间轮实现

接口定义

Netty 的实现自定义了一个超时器的接口io.netty.util.Timer，其方法如下:

1. publicinterfaceTimer
2. {
3. //新增一个延时任务，入参为定时任务TimerTask，和对应的延迟时间
4. TimeoutnewTimeout(TimerTasktask,longdelay,TimeUnitunit);
5. //停止时间轮的运行，并且返回所有未被触发的延时任务
6. Set<Timeout>stop();
7. }
8. publicinterfaceTimeout
9. {
10. Timertimer();
11. TimerTasktask();
12. booleanisExpired();
13. booleanisCancelled();
14. booleancancel();
15. }

Timeout接口是对延迟任务的一个封装，其接口方法说明其实现内部需要维持该延迟任务的状态。后续我们分析其实现内部代码时可以更容易的看到。

Timer接口有唯一实现HashedWheelTimer。首先来看其构造方法，如下：

1. publicHashedWheelTimer(ThreadFactorythreadFactory,longtickDuration,TimeUnitunit,intticksPerWheel,booleanleakDetection,longmaxPendingTimeouts)
2. {
3. //省略代码，省略参数非空检查内容。
4. wheel=createWheel(ticksPerWheel);
5. mask=wheel.length-1;
6. //省略代码，省略槽位时间范围检查，避免溢出以及小于1毫秒。
7. workerThread=threadFactory.newThread(worker);
8. //省略代码，省略资源泄漏追踪设置以及时间轮实例个数检查
9. }

mask 的设计和HashMap一样，通过限制数组的大小为2的次方，利用位运算来替代取模运算，提高性能。

构建循环数组

首先是方法createWheel，用于创建时间轮的核心数据结构，循环数组。来看下其方法内容

1. privatestaticHashedWheelBucket[]createWheel(intticksPerWheel)
2. {
3. //省略代码，确认ticksPerWheel处于正确的区间
4. //将ticksPerWheel规范化为2的次方幂大小。
5. ticksPerWheel=normalizeTicksPerWheel(ticksPerWheel);
6. HashedWheelBucket[]wheel=newHashedWheelBucket[ticksPerWheel];
7. for(inti=0;i<wheel.length;i++)
8. {
9. wheel[i]=newHashedWheelBucket();
10. }
11. returnwheel;
12. }

数组的长度为 2 的次方幂方便进行求商和取余计算。

HashedWheelBucket内部存储着由HashedWheelTimeout节点构成的双向链表，并且存储着链表的头节点和尾结点，方便于任务的提取和插入。

新增延迟任务

方法HashedWheelTimer#newTimeout用于新增延迟任务，下面来看下代码:

1. publicTimeoutnewTimeout(TimerTasktask,longdelay,TimeUnitunit)
2. {
3. //省略代码，用于参数检查
4. start();
5. longdeadline=System.nanoTime()+unit.toNanos(delay)-startTime;
6. if(delay>0&&deadline<0)
7. {
8. deadline=Long.MAX_VALUE;
9. }
10. HashedWheelTimeouttimeout=newHashedWheelTimeout(this,task,deadline);
11. timeouts.add(timeout);
12. returntimeout;
13. }

可以看到任务并没有直接添加到时间轮中，而是先入了一个 mpsc 队列，我简单说下 mpsc【多生产者单一消费者队列】 是 JCTools 中的并发队列，用在多个生产者可同时访问队列，但只有一个消费者会访问队列的情况。，采用这个模式主要出于提升并发性能考虑，因为这个队列只有线程workerThread会进行任务提取操作。

工作线程如何执行

1. publicvoidrun()
2. {
3. {//代码块①
4. startTime=System.nanoTime();
5. if(startTime==0)
6. {
7. //使用startTime==0作为线程进入工作状态模式标识，因此这里重新赋值为1
8. startTime=1;
9. }
10. //通知外部初始化工作线程的线程，工作线程已经启动完毕
11. startTimeInitialized.countDown();
12. }
13. {//代码块②
14. do{
15. finallongdeadline=waitForNextTick();
16. if(deadline>0)
17. {
18. intidx=(int)(tick&mask);
19. processCancelledTasks();
20. HashedWheelBucketbucket=wheel[idx];
21. transferTimeoutsToBuckets();
22. bucket.expireTimeouts(deadline);
23. tick++;
24. }
25. }while(WORKER_STATE_UPDATER.get(HashedWheelTimer.this)==WORKER_STATE_STARTED);
26. }
27. {//代码块③
28. for(HashedWheelBucketbucket:wheel)
29. {
30. bucket.clearTimeouts(unprocessedTimeouts);
31. }
32. for(;;)
33. {
34. HashedWheelTimeouttimeout=timeouts.poll();
35. if(timeout==null)
36. {
37. break;
38. }
39. if(!timeout.isCancelled())
40. {
41. unprocessedTimeouts.add(timeout);
42. }
43. }
44. processCancelledTasks();
45. }
46. }

看 waitForNextTick，是如何得到下一次执行时间的。

1. privatelongwaitForNextTick()
2. {
3. longdeadline=tickDuration*(tick+1);//计算下一次需要检查的时间
4. for(;;)
5. {
6. finallongcurrentTime=System.nanoTime()-startTime;
7. longsleepTimeMs=(deadline-currentTime+999999)/1000000;
8. if(sleepTimeMs<=0)//说明时间已经到了
9. {
10. if(currentTime==Long.MIN_VALUE)
11. {
12. return-Long.MAX_VALUE;
13. }
14. else
15. {
16. returncurrentTime;
17. }
18. }
19. //windows下有bugsleep必须是10的倍数
20. if(PlatformDependent.isWindows())
21. {
22. sleepTimeMs=sleepTimeMs/10*10;
23. }
24. try
25. {
26. Thread.sleep(sleepTimeMs);//等待时间到来
27. }
28. catch(InterruptedExceptionignored)
29. {
30. if(WORKER_STATE_UPDATER.get(HashedWheelTimer.this)==WORKER_STATE_SHUTDOWN)
31. {
32. returnLong.MIN_VALUE;
33. }
34. }
35. }
36. }

简单的说就是通过 tickDuration 和此时已经滴答的次数算出下一次需要检查的时间，时候未到就sleep等着。

任务如何入槽的。

1. privatevoidtransferTimeoutsToBuckets(){
2. //最多处理100000怕任务延迟
3. for(inti=0;i<100000;++i){
4. //从队列里面拿出任务呢
5. HashedWheelTimer.HashedWheelTimeouttimeout=(HashedWheelTimer.HashedWheelTimeout)HashedWheelTimer.this.timeouts.poll();
6. if(timeout==null){
7. break;
8. }
10. if(timeout.state()!=1){
11. longcalculated=timeout.deadline/HashedWheelTimer.this.tickDuration;
12. //计算排在第几轮
13. timeout.remainingRounds=(calculated-this.tick)/(long)HashedWheelTimer.this.wheel.length;
14. longticks=Math.max(calculated,this.tick);
15. //计算放在哪个槽中
16. intstopIndex=(int)(ticks&(long)HashedWheelTimer.this.mask);
17. HashedWheelTimer.HashedWheelBucketbucket=HashedWheelTimer.this.wheel[stopIndex];
18. //入槽，就是链表入队列
19. bucket.addTimeout(timeout);
20. }
21. }
23. }

如何执行的

1. publicvoidexpireTimeouts(longdeadline){
2. HashedWheelTimer.HashedWheelTimeoutnext;
3. //拿到槽的链表头部
4. for(HashedWheelTimer.HashedWheelTimeouttimeout=this.head;timeout!=null;timeout=next){
5. booleanremove=false;
6. if(timeout.remainingRounds<=0L){//如果到这轮l
7. if(timeout.deadline>deadline){
8. thrownewIllegalStateException(String.format("timeout.deadline(%d)>deadline(%d)",timeout.deadline,deadline));
9. }
11. timeout.expire();//执行
12. remove=true;
13. }elseif(timeout.isCancelled()){
14. remove=true;
15. }else{
16. --timeout.remainingRounds;//轮数-1
17. }
19. next=timeout.next;//继续下一任务
20. if(remove){
21. this.remove(timeout);//移除完成的任务
22. }
23. }
24. }

就是通过轮数和时间双重判断，执行完了移除任务。

小结一下

总体上看 Netty 的实现就是上文说的时间轮通过轮数的实现，完全一致。可以看出时间精度由 TickDuration 把控，并且工作线程的除了处理执行到时的任务还做了其他操作，因此任务不一定会被精准的执行。

而且任务的执行如果不是新起一个线程，或者将任务扔到线程池执行，那么耗时的任务会阻塞下个任务的执行。

并且会有很多无用的 tick 推进，例如 TickDuration 为1秒，此时就一个延迟350秒的任务，那就是有349次无用的操作。出现空推。

但是从另一面来看，如果任务都执行很快(当然你也可以异步执行)，并且任务数很多，通过分批执行，并且增删任务的时间复杂度都是O(1)来说。时间轮还是比通过优先队列实现的延时任务来的合适些。

Kafka 中的时间轮

上面我们说到 Kafka 中的时间轮是多层次时间轮实现，总的而言实现和上述说的思路一致。不过细节有些不同，并且做了点优化。

先看看添加任务的方法。在添加的时候就设置任务执行的绝对时间。

Kafka 中的时间轮

上面我们说到 Kafka 中的时间轮是多层次时间轮实现，总的而言实现和上述说的思路一致。不过细节有些不同，并且做了点优化。

先看看添加任务的方法。在添加的时候就设置任务执行的绝对时间。

1. defadd(timerTaskEntry:TimerTaskEntry):Boolean={
2. valexpiration=timerTaskEntry.expirationMs
4. if(timerTaskEntry.cancelled){
5. //Cancelled
6. false
7. }elseif(expiration<currentTime+tickMs){
8. //如果已经到期返回false
9. //Alreadyexpired
10. false
11. }elseif(expiration<currentTime+interval){//如果在本层范围内
12. //Putinitsownbucket
13. valvirtualId=expiration/tickMs
14. valbucket=buckets((virtualId%wheelSize.toLong).toInt)//计算槽位
15. bucket.add(timerTaskEntry)//添加到槽内双向链表中
17. //Setthebucketexpirationtime
18. if(bucket.setExpiration(virtualId*tickMs)){//更新槽时间
19. //Thebucketneedstobeenqueuedbecauseitwasanexpiredbucket
20. //Weonlyneedtoenqueuethebucketwhenitsexpirationtimehaschanged,i.e.thewheelhasadvanced
21. //andthepreviousbucketsgetsreused;furthercallstosettheexpirationwithinthesamewheelcycle
22. //willpassinthesamevalueandhencereturnfalse,thusthebucketwiththesameexpirationwillnot
23. //beenqueuedmultipletimes.
24. queue.offer(bucket)//将槽加入DelayQueue，由DelayQueue来推进执行
25. }
26. true
27. }else{
28. //如果超过本层能表示的延迟时间，则将任务添加到上层。这里看到上层是按需创建的。
29. //Outoftheinterval.Putitintotheparenttimer
30. if(overflowWheel==null)addOverflowWheel()
31. overflowWheel.add(timerTaskEntry)
32. }
33. }

那么时间轮是如何推动的呢?Netty 中是通过固定的时间间隔扫描，时候未到就等待来进行时间轮的推动。上面我们分析到这样会有空推进的情况。

而 Kafka 就利用了空间换时间的思想，通过 DelayQueue，来保存每个槽，通过每个槽的过期时间排序。这样拥有最早需要执行任务的槽会有优先获取。如果时候未到，那么 delayQueue.poll 就会阻塞着，这样就不会有空推进的情况发送。

我们来看下推进的方法。

1. defadvanceClock(timeoutMs:Long):Boolean={
2. //从延迟队列中获取槽
3. varbucket=delayQueue.poll(timeoutMs,TimeUnit.MILLISECONDS)
4. if(bucket!=null){
5. writeLock.lock()
6. try{
7. while(bucket!=null){
8. //更新每层时间轮的currentTime
9. timingWheel.advanceClock(bucket.getExpiration())
10. //因为更新了currentTime，进行一波任务的重新插入，来实现任务时间轮的降级
11. bucket.flush(reinsert)
12. //获取下一个槽
13. bucket=delayQueue.poll()
14. }
15. }finally{
16. writeLock.unlock()
17. }
18. true
19. }else{
20. false
21. }
22. }
24. //Trytoadvancetheclock
25. defadvanceClock(timeMs:Long):Unit={
26. if(timeMs>=currentTime+tickMs){
27. //必须是tickMs整数倍
28. currentTime=timeMs-(timeMs%tickMs)
29. //推动上层时间轮也更新currentTime
30. //Trytoadvancetheclockoftheoverflowwheelifpresent
31. if(overflowWheel!=null)overflowWheel.advanceClock(currentTime)
32. }
33. }

从上面的 add 方法我们知道每次对比都是根据expiration < currentTime + interval 来进行对比的，而advanceClock 就是用来推进更新 currentTime 的。

小结一下

Kafka 用了多层次时间轮来实现，并且是按需创建时间轮，采用任务的绝对时间来判断延期，并且对于每个槽(槽内存放的也是任务的双向链表)都会维护一个过期时间，利用 DelayQueue 来对每个槽的过期时间排序，来进行时间的推进，防止空推进的存在。

每次推进都会更新 currentTime 为当前时间戳，当然做了点微调使得 currentTime 是 tickMs 的整数倍。并且每次推进都会把能降级的任务重新插入降级。

可以看到这里的 DelayQueue 的元素是每个槽，而不是任务，因此数量就少很多了，这应该是权衡了对于槽操作的延时队列的时间复杂度与空推进的影响。

模拟kafka的时间轮实现java版

定时器

1. publicclassTimer{
3. /**
4. *底层时间轮
5. */
6. privateTimeWheeltimeWheel;
8. /**
9. *一个Timer只有一个delayQueue
10. */
11. privateDelayQueue<TimerTaskList>delayQueue=newDelayQueue<>();
13. /**
14. *过期任务执行线程
15. */
16. privateExecutorServiceworkerThreadPool;
18. /**
19. *轮询delayQueue获取过期任务线程
20. */
21. privateExecutorServicebossThreadPool;
23. /**
24. *构造函数
25. */
26. publicTimer(){
27. timeWheel=newTimeWheel(1000,2,System.currentTimeMillis(),delayQueue);
28. workerThreadPool=Executors.newFixedThreadPool(100);
29. bossThreadPool=Executors.newFixedThreadPool(1);
30. //20ms获取一次过期任务
31. bossThreadPool.submit(()->{
32. while(true){
33. this.advanceClock(1000);
34. }
35. });
36. }
38. /**
39. *添加任务
40. */
41. publicvoidaddTask(TimerTasktimerTask){
42. //添加失败任务直接执行
43. if(!timeWheel.addTask(timerTask)){
44. workerThreadPool.submit(timerTask.getTask());
45. }
46. }
48. /**
49. *获取过期任务
50. */
51. privatevoidadvanceClock(longtimeout){
52. try{
53. TimerTaskListtimerTaskList=delayQueue.poll(timeout,TimeUnit.MILLISECONDS);
54. if(timerTaskList!=null){
56. //推进时间
57. timeWheel.advanceClock(timerTaskList.getExpiration());
58. //执行过期任务（包含降级操作）
59. timerTaskList.flush(this::addTask);
60. }
61. }catch(Exceptione){
62. e.printStackTrace();
63. }
64. }
65. }

任务

1. publicclassTimerTask{
3. /**
4. *延迟时间
5. */
6. privatelongdelayMs;
8. /**
9. *任务
10. */
11. privateMyThreadtask;
13. /**
14. *时间槽
15. */
16. protectedTimerTaskListtimerTaskList;
18. /**
19. *下一个节点
20. */
21. protectedTimerTasknext;
23. /**
24. *上一个节点
25. */
26. protectedTimerTaskpre;
28. /**
29. *描述
30. */
31. publicStringdesc;
33. publicTimerTask(longdelayMs,MyThreadtask){
34. this.delayMs=System.currentTimeMillis()+delayMs;
35. this.task=task;
36. this.timerTaskList=null;
37. this.next=null;
38. this.pre=null;
39. }
41. publicMyThreadgetTask(){
42. returntask;
43. }
45. publiclonggetDelayMs(){
46. returndelayMs;
47. }
49. @Override
50. publicStringtoString(){
51. returndesc;
52. }
53. }

时间槽

1. publicclassTimerTaskListimplementsDelayed{
3. /**
4. *过期时间
5. */
6. privateAtomicLongexpiration=newAtomicLong(-1L);
8. /**
9. *根节点
10. */
11. privateTimerTaskroot=newTimerTask(-1L,null);
13. {
14. root.pre=root;
15. root.next=root;
16. }
18. /**
19. *设置过期时间
20. */
21. publicbooleansetExpiration(longexpire){
22. returnexpiration.getAndSet(expire)!=expire;
23. }
25. /**
26. *获取过期时间
27. */
28. publiclonggetExpiration(){
29. returnexpiration.get();
30. }
32. /**
33. *新增任务
34. */
35. publicvoidaddTask(TimerTasktimerTask){
36. synchronized(this){
37. if(timerTask.timerTaskList==null){
38. timerTask.timerTaskList=this;
39. TimerTasktail=root.pre;
40. timerTask.next=root;
41. timerTask.pre=tail;
42. tail.next=timerTask;
43. root.pre=timerTask;
44. }
45. }
46. }
48. /**
49. *移除任务
50. */
51. publicvoidremoveTask(TimerTasktimerTask){
52. synchronized(this){
53. if(timerTask.timerTaskList.equals(this)){
54. timerTask.next.pre=timerTask.pre;
55. timerTask.pre.next=timerTask.next;
56. timerTask.timerTaskList=null;
57. timerTask.next=null;
58. timerTask.pre=null;
59. }
60. }
61. }
63. /**
64. *重新分配
65. */
66. publicsynchronizedvoidflush(Consumer<TimerTask>flush){
67. TimerTasktimerTask=root.next;
68. while(!timerTask.equals(root)){
69. this.removeTask(timerTask);
70. flush.accept(timerTask);
71. timerTask=root.next;
72. }
73. expiration.set(-1L);
74. }
76. @Override
77. publiclonggetDelay(TimeUnitunit){
78. returnMath.max(0,unit.convert(expiration.get()-System.currentTimeMillis(),TimeUnit.MILLISECONDS));
79. }
81. @Override
82. publicintcompareTo(Delayedo){
83. if(oinstanceofTimerTaskList){
84. returnLong.compare(expiration.get(),((TimerTaskList)o).expiration.get());
85. }
86. return0;
87. }
88. }

时间轮

1. publicclassTimeWheel{
3. /**
4. *一个时间槽的范围
5. */
6. privatelongtickMs;
8. /**
9. *时间轮大小
10. */
11. privateintwheelSize;
13. /**
14. *时间跨度
15. */
16. privatelonginterval;
18. /**
19. *时间槽
20. */
21. privateTimerTaskList[]timerTaskLists;
23. /**
24. *当前时间
25. */
26. privatelongcurrentTime;
28. /**
29. *上层时间轮
30. */
31. privatevolatileTimeWheeloverflowWheel;
33. /**
34. *一个Timer只有一个delayQueue
35. */
36. privateDelayQueue<TimerTaskList>delayQueue;
38. publicTimeWheel(longtickMs,intwheelSize,longcurrentTime,DelayQueue<TimerTaskList>delayQueue){
39. this.currentTime=currentTime;
40. this.tickMs=tickMs;
41. this.wheelSize=wheelSize;
42. this.interval=tickMs*wheelSize;
43. this.timerTaskLists=newTimerTaskList[wheelSize];
44. //currentTime为tickMs的整数倍这里做取整操作
45. this.currentTime=currentTime-(currentTime%tickMs);
46. this.delayQueue=delayQueue;
47. for(inti=0;i<wheelSize;i++){
48. timerTaskLists[i]=newTimerTaskList();
49. }
50. }
52. /**
53. *创建或者获取上层时间轮
54. */
55. privateTimeWheelgetOverflowWheel(){
56. if(overflowWheel==null){
57. synchronized(this){
58. if(overflowWheel==null){
59. overflowWheel=newTimeWheel(interval,wheelSize,currentTime,delayQueue);
60. }
61. }
62. }
63. returnoverflowWheel;
64. }
66. /**
67. *添加任务到时间轮
68. */
69. publicbooleanaddTask(TimerTasktimerTask){
70. longexpiration=timerTask.getDelayMs();
71. //过期任务直接执行
72. if(expiration<currentTime+tickMs){
73. returnfalse;
74. }elseif(expiration<currentTime+interval){
75. //当前时间轮可以容纳该任务加入时间槽
76. LongvirtualId=expiration/tickMs;
77. intindex=(int)(virtualId%wheelSize);
78. System.out.println("tickMs:"+tickMs+"------index:"+index+"------expiration:"+expiration);
79. TimerTaskListtimerTaskList=timerTaskLists[index];
80. timerTaskList.addTask(timerTask);
81. if(timerTaskList.setExpiration(virtualId*tickMs)){
82. //添加到delayQueue中
83. delayQueue.offer(timerTaskList);
84. }
85. }else{
86. //放到上一层的时间轮
87. TimeWheeltimeWheel=getOverflowWheel();
88. timeWheel.addTask(timerTask);
89. }
90. returntrue;
91. }
93. /**
94. *推进时间
95. */
96. publicvoidadvanceClock(longtimestamp){
97. if(timestamp>=currentTime+tickMs){
98. currentTime=timestamp-(timestamp%tickMs);
99. if(overflowWheel!=null){
100. //推进上层时间轮时间
101. System.out.println("推进上层时间轮时间time="+System.currentTimeMillis());
102. this.getOverflowWheel().advanceClock(timestamp);
103. }
104. }
105. }
106. }

我们来模拟一个请求，超时和不超时的情况

首先定义一个Mythread 类，用于设置任务超时的值。

1. publicclassMyThreadimplementsRunnable{
2. CompletableFuture<String>cf;
3. publicMyThread(CompletableFuture<String>cf){
4. this.cf=cf;
5. }
6. publicvoidrun(){
7. if(!cf.isDone()){
8. cf.complete("超时");
9. }
10. }
11. }

模拟超时

1. publicstaticvoidmain(String[]args)throwsException{
2. Timertimer=newTimer();
3. CompletableFuture<String>base=CompletableFuture.supplyAsync(()->{
4. try{
5. Thread.sleep(3000);
6. }catch(InterruptedExceptione){
7. e.printStackTrace();
8. }
9. return"正常返回";
10. });
11. TimerTasktimerTask2=newTimerTask(1000,newMyThread(base));
12. timer.addTask(timerTask2);
13. System.out.println("base.get==="+base.get());
14. }

模拟正常返回

1. publicstaticvoidmain(String[]args)throwsException{
2. Timertimer=newTimer();
3. CompletableFuture<String>base=CompletableFuture.supplyAsync(()->{
4. try{
5. Thread.sleep(300);
6. }catch(InterruptedExceptione){
7. e.printStackTrace();
8. }
9. return"正常返回";
10. });
11. TimerTasktimerTask2=newTimerTask(2000,newMyThread(base));
12. timer.addTask(timerTask2);
13. System.out.println("base.get==="+base.get());
14. }

原文地址：https://mp.weixin.qq.com/s/8uCN4OL3S1aoT8ff_2QXhQ

如果您对该产品感兴趣，请填写办理（客服微信：xiaoxiongyidong）

您的姓名 *

身份证号码 *

手机号码 *

选择地区 *

详细地址 *

留言备注 *