Go语言中时间轮的实现

介绍

简单时间轮

taskList

timewheel

timewheelAdd9S

层级时间轮

timewheellevel2

代码实现

• 时间轮的时间格中每个链表会有一个 root 节点用于简化边界条件。它是一个附加的链表节点，该节点作为第一个节点，它的值域中并不存储任何东西，只是为了操作的方便而引入的；
• 除了第一层时间轮，其余高层时间轮的起始时间（startMs）都设置为创建此层时间轮时前面第一轮的 currentTime。每一层的 currentTime 都必须是 tickMs 的整数倍，如果不满足则会将 currentTime 修剪为 tickMs 的整数倍。修剪方法为：currentTime = startMs - (startMs % tickMs)；
• Kafka 中的定时器只需持有 TimingWheel 的第一层时间轮的引用，并不会直接持有其他高层的时间轮，但每一层时间轮都会有一个引用（overflowWheel）指向更高一层的应用；
• Kafka 中的定时器使用了 DelayQueue 来协助推进时间轮。在操作中会将每个使用到的时间格中每个链表都加入 DelayQueue，DelayQueue 会根据时间轮对应的过期时间 expiration 来排序，最短 expiration 的任务会被排在 DelayQueue 的队头，通过单独线程来获取 DelayQueue 中到期的任务；

结构体

type TimingWheel struct { // 时间跨度,单位是毫秒 tick      int64 // in milliseconds // 时间轮个数 wheelSize int64 // 总跨度 interval    int64 // in milliseconds // 当前指针指向时间 currentTime int64 // in milliseconds // 时间格列表 buckets     []*bucket // 延迟队列 queue       *delayqueue.DelayQueue  // 上级的时间轮引用 overflowWheel unsafe.Pointer // type: *TimingWheel
 exitC     chan struct{} waitGroup waitGroupWrapper}

type bucket struct { // 任务的过期时间 expiration int64
 mu     sync.Mutex // 相同过期时间的任务队列 timers *list.List}

type Timer struct {  // 到期时间 expiration int64 // in milliseconds  // 要被执行的具体任务 task       func() // Timer所在bucket的指针 b unsafe.Pointer // type: *bucket // bucket列表中对应的元素 element *list.Element}

Group 37

初始化时间轮

func main() { tw := timingwheel.NewTimingWheel(time.Second, 10) tw.Start() }func NewTimingWheel(tick time.Duration, wheelSize int64) *TimingWheel {  // 将传入的tick转化成毫秒 tickMs := int64(tick / time.Millisecond)  // 如果小于零，那么panic if tickMs <= 0 {  panic(errors.New("tick must be greater than or equal to 1ms")) } // 设置开始时间 startMs := timeToMs(time.Now().UTC()) // 初始化TimingWheel return newTimingWheel(  tickMs,  wheelSize,  startMs,  delayqueue.New(int(wheelSize)), )}func newTimingWheel(tickMs int64, wheelSize int64, startMs int64, queue *delayqueue.DelayQueue) *TimingWheel {  // 初始化buckets的大小 buckets := make([]*bucket, wheelSize) for i := range buckets {  buckets[i] = newBucket() }  // 实例化TimingWheel return &TimingWheel{  tick:        tickMs,  wheelSize:   wheelSize,    // currentTime必须是tickMs的倍数，所以这里使用truncate进行修剪  currentTime: truncate(startMs, tickMs),  interval:    tickMs * wheelSize,  buckets:     buckets,  queue:       queue,  exitC:       make(chan struct{}), }}

启动时间轮

func (tw *TimingWheel) Start() { // Poll会执行一个无限循环，将到期的元素放入到queue的C管道中 tw.waitGroup.Wrap(func() {  tw.queue.Poll(tw.exitC, func() int64 {   return timeToMs(time.Now().UTC())  }) }) // 开启无限循环获取queue中C的数据 tw.waitGroup.Wrap(func() {  for {   select {   // 从队列里面出来的数据都是到期的bucket   case elem := <-tw.queue.C:    b := elem.(*bucket)    // 时间轮会将当前时间 currentTime 往前移动到 bucket的到期时间    tw.advanceClock(b.Expiration())    // 取出bucket队列的数据，并调用addOrRun方法执行    b.Flush(tw.addOrRun)   case <-tw.exitC:    return   }  } })}

func (tw *TimingWheel) advanceClock(expiration int64) { currentTime := atomic.LoadInt64(&tw.currentTime) // 过期时间大于等于（当前时间+tick） if expiration >= currentTime+tw.tick {  // 将currentTime设置为expiration，从而推进currentTime  currentTime = truncate(expiration, tw.tick)  atomic.StoreInt64(&tw.currentTime, currentTime)
  // Try to advance the clock of the overflow wheel if present  // 如果有上层时间轮，那么递归调用上层时间轮的引用  overflowWheel := atomic.LoadPointer(&tw.overflowWheel)  if overflowWheel != nil {   (*TimingWheel)(overflowWheel).advanceClock(currentTime)  } }}

func (b *bucket) Flush(reinsert func(*Timer)) { var ts []*Timer
 b.mu.Lock() // 循环获取bucket队列节点 for e := b.timers.Front(); e != nil; {  next := e.Next()
  t := e.Value.(*Timer)  // 将头节点移除bucket队列  b.remove(t)  ts = append(ts, t)
  e = next } b.mu.Unlock()
 b.SetExpiration(-1) // TODO: Improve the coordination with b.Add()
 for _, t := range ts {  reinsert(t) }}

func (tw *TimingWheel) addOrRun(t *Timer) { // 如果已经过期，那么直接执行 if !tw.add(t) {   // 异步执行定时任务  go t.task() }}

timewheel_start

1. start 方法会启动一个 goroutines 调用 poll 来处理 DelayQueue 中到期的数据，并将数据放入到管道 C 中；
2. start 方法启动第二个 goroutines 方法会循环获取 DelayQueue 中管道C的数据，管道 C 中实际上存放的是一个 bucket，然后遍历bucket的timers列表，如果任务已经到期，那么异步执行，没有到期则重新放入到 DelayQueue 中。

add task

func main() { tw := timingwheel.NewTimingWheel(time.Second, 10) tw.Start()  // 添加任务 tw.AfterFunc(time.Second*15, func() {  fmt.Println("The timer fires")  exitC <- time.Now().UTC() })}

func (tw *TimingWheel) AfterFunc(d time.Duration, f func()) *Timer { t := &Timer{  expiration: timeToMs(time.Now().UTC().Add(d)),  task:       f, } tw.addOrRun(t) return t}

func (tw *TimingWheel) add(t *Timer) bool { currentTime := atomic.LoadInt64(&tw.currentTime) // 已经过期 if t.expiration < currentTime+tw.tick {  // Already expired  return false //  到期时间在第一层环内 } else if t.expiration < currentTime+tw.interval {  // Put it into its own bucket  // 获取时间轮的位置  virtualID := t.expiration / tw.tick  b := tw.buckets[virtualID%tw.wheelSize]  // 将任务放入到bucket队列中  b.Add(t)   // 如果是相同的时间，那么返回false，防止被多次插入到队列中  if b.SetExpiration(virtualID * tw.tick) {    // 将该bucket加入到延迟队列中   tw.queue.Offer(b, b.Expiration())  }
  return true } else {  // Out of the interval. Put it into the overflow wheel  // 如果放入的到期时间超过第一层时间轮，那么放到上一层中去  overflowWheel := atomic.LoadPointer(&tw.overflowWheel)  if overflowWheel == nil {   atomic.CompareAndSwapPointer(    &tw.overflowWheel,    nil,    // 需要注意的是，这里tick变成了interval    unsafe.Pointer(newTimingWheel(     tw.interval,     tw.wheelSize,     currentTime,     tw.queue,    )),   )   overflowWheel = atomic.LoadPointer(&tw.overflowWheel)  }  // 往上递归  return (*TimingWheel)(overflowWheel).add(t) }}

Reference

• https://github.com/RussellLuo/timingwheel
• https://zhuanlan.zhihu.com/p/121483218
• https://github.com/apache/kafka/tree/3cdc78e6bb1f83973a14ce1550fe3874f7348b05/core/src/main/scala/kafka/utils/timer

☞股价狂涨 500 亿，小米手机业务与造车可否兼得？
☞RISC-V 没你想象的那么好
☞30 周岁的 Python，“虐”我 20 年
☞谷歌称远程办公影响工作效率，2021年回归线下办公