基于 golang 从零到一实现时间轮算法 (二)

Go实现单机版时间轮

上一章介绍了时间轮的相关概念，接下来我们会使用 golang 标准库的定时器工具 time ticker 结合环状数组的设计思路，实现一个单机版的单级时间轮。
首先我们先运行一下下面的源码，看一下如何使用。

https://github.com/xiaoxuxiansheng/timewheel

package mainimport ("container/list""fmt""sync""time"
)type taskElement struct {task  func()pos   intcycle intkey   string
}type TimeWheel struct {sync.Onceinterval     time.Durationticker       *time.Tickerstopc        chan struct{}addTaskCh    chan *taskElementremoveTaskCh chan stringslots        []*list.ListcurSlot      intkeyToETask   map[string]*list.Element
}func NewTimeWheel(slotNum int, interval time.Duration) *TimeWheel {if slotNum <= 0 {slotNum = 10}if interval <= 0 {interval = time.Second}t := TimeWheel{interval:     interval,ticker:       time.NewTicker(interval),stopc:        make(chan struct{}),keyToETask:   make(map[string]*list.Element),slots:        make([]*list.List, 0, slotNum),addTaskCh:    make(chan *taskElement),removeTaskCh: make(chan string),}for i := 0; i < slotNum; i++ {t.slots = append(t.slots, list.New())}go t.run()return &t
}func (t *TimeWheel) Stop() {t.Do(func() {t.ticker.Stop()close(t.stopc)})
}func (t *TimeWheel) AddTask(key string, task func(), executeAt time.Time) {pos, cycle := t.getPosAndCircle(executeAt)t.addTaskCh <- &taskElement{pos:   pos,cycle: cycle,task:  task,key:   key,}
}func (t *TimeWheel) RemoveTask(key string) {t.removeTaskCh <- key
}func (t *TimeWheel) run() {defer func() {if err := recover(); err != nil {// ...}}()for {select {case <-t.stopc:returncase <-t.ticker.C:t.tick()case task := <-t.addTaskCh:t.addTask(task)case removeKey := <-t.removeTaskCh:t.removeTask(removeKey)}}
}func (t *TimeWheel) tick() {list := t.slots[t.curSlot]defer t.circularIncr()t.execute(list)
}func (t *TimeWheel) execute(l *list.List) {// 遍历每个 listfor e := l.Front(); e != nil; {taskElement, _ := e.Value.(*taskElement)if taskElement.cycle > 0 {taskElement.cycle--e = e.Next()continue}// 执行任务go func() {defer func() {if err := recover(); err != nil {// ...}}()taskElement.task()}()// 执行任务后，从时间轮中删除next := e.Next()l.Remove(e)delete(t.keyToETask, taskElement.key)e = next}
}func (t *TimeWheel) getPosAndCircle(executeAt time.Time) (int, int) {delay := int(time.Until(executeAt))cycle := delay / (len(t.slots) * int(t.interval))pos := (t.curSlot + delay/int(t.interval)) % len(t.slots)return pos, cycle
}func (t *TimeWheel) addTask(task *taskElement) {list := t.slots[task.pos]if _, ok := t.keyToETask[task.key]; ok {t.removeTask(task.key)}eTask := list.PushBack(task)t.keyToETask[task.key] = eTask
}func (t *TimeWheel) removeTask(key string) {eTask, ok := t.keyToETask[key]if !ok {return}delete(t.keyToETask, key)task, _ := eTask.Value.(*taskElement)_ = t.slots[task.pos].Remove(eTask)
}func (t *TimeWheel) circularIncr() {t.curSlot = (t.curSlot + 1) % len(t.slots)
}func main() {timeWheel := NewTimeWheel(10, 500*time.Millisecond)defer timeWheel.Stop()fmt.Println(time.Now())timeWheel.AddTask("test1", func() {fmt.Printf("test1, %v\n", time.Now())}, time.Now().Add(time.Second))timeWheel.AddTask("test2", func() {fmt.Printf("test2, %v\n", time.Now())}, time.Now().Add(5*time.Second))timeWheel.AddTask("test2", func() {fmt.Printf("test2, %v\n", time.Now())}, time.Now().Add(3*time.Second))<-time.After(10 * time.Second)
}

运行结果如下：

2023-11-03 13:02:57.042834 +0800 CST m=+0.000173292
test1, 2023-11-03 13:02:58.043555 +0800 CST m=+1.000891376
test2, 2023-11-03 13:03:00.043567 +0800 CST m=+3.000897126

结果说明，首先添加test1任务定时1秒钟，test2任务定时5秒钟，但后续修改了test2定时为3秒钟，所以输出test1和test2的时间差为2秒钟。

---

数据结构

在对时间轮的类定义中，核心字段如下图所示：

type TimeWheel struct {sync.Onceinterval     time.Durationticker       *time.Tickerstopc        chan struct{}addTaskCh    chan *taskElementremoveTaskCh chan stringslots        []*list.ListcurSlot      intkeyToETask   map[string]*list.Element
}

在几个核心字段中：

• slots——类似于时钟的表盘
• curSlot——类似于时钟的指针
• ticker 是使用 golang标准库的定时器工具，类似于驱动指针运转的齿轮

在创建时间轮实例时，会通过一个异步的常驻 goroutine 执行定时任务的检索、添加、删除等操作，并通过几个 channel 进行 goroutine 的执行逻辑和生命周期的控制：

• stopc：用于停止 goroutine
• addTaskCh：用于接收创建定时器指令
• removeTaskCh：用于接收删除定时任务的指令

此处有几个技术细节需要提及：

首先：所谓环状数组指的是逻辑意义上的. 在实际的实现过程中，会通过一个定长数组结合循环遍历的方式，来实现这个逻辑意义上的“环状”性质.（有点类似于上一章提到的cycle）

其次：数组每一轮能表达的时间范围是固定的. 每当在添加添加一个定时任务时，需要根据其延迟的相对时长推算出其所处的 slot 位置，其中可能跨遍历轮次的情况，这时候需要额外通过定时任务中的 cycle 字段来记录这一信息，避免定时任务被提前执行.

最后：时间轮中一个 slot 可能需要挂载多笔定时任务，因此针对每个 slot，需要采用 golang 标准库 container/list 中实现的双向链表进行定时任务数据的存储.

定时任务

我们现在先看一笔任务的结构体介绍：

// 封装了一笔定时任务的明细信息
type taskElement struct {// 内聚了定时任务执行逻辑的闭包函数task  func()// 定时任务挂载在环状数组中的索引位置pos   int// 定时任务的延迟轮次. 指的是 curSlot 指针还要扫描过环状数组多少轮，才满足执行该任务的条件cycle int// 定时任务的唯一标识键key   string
}

• task func(): 这是一个函数类型的字段，它引用了一个闭包。闭包是一种匿名函数，能够捕获到其外部作用域中的变量。在这里，task字段代表着定时任务的执行逻辑本身。当定时器触发时，这个闭包会被执行。这样设计可以让taskElement持有执行任务所需要进行的任何操作，使任务逻辑高度内聚和独立。

• pos int: 该字段表示任务在环形数组（通常用于实现时间轮定时器）中的位置索引。环形数组是时间轮算法中的一种数据结构，用来表示时间的流逝。pos就是这个任务在这个环中的具体位置，当时间轮的指针指向这个位置时，就意味着这个taskElement代表的定时任务可能需要被执行。

• cycle int: 在时间轮算法中，cycle用于表示任务延迟的轮次数。时间轮有一个当前指针curSlot，每当curSlot遍历一次完整的环形数组，所有任务的cycle值都会减1。一个任务的cycle值指示了curSlot需要再经过多少完整的遍历，该任务才会被执行。当cycle为0时，表示定时任务在当前轮次达到了执行条件。

• key string: 这个字段是每个定时任务的唯一标识。key的存在允许任务在全局范围内被唯一标识和引用。这意味着你可以使用这个key来查询或者操作特定的定时任务，比如更新任务的延迟时间、取消任务或者是在任务被执行之前获取任务的状态。

综上所述，taskElement结构体将一个定时任务的执行逻辑、在时间轮中的位置、剩余的延迟轮次以及唯一标识符组合在一起，为定时任务的调度提供了必要的信息。

构造器

在创建时间轮的构造器函数中，需要传入两个入参：

• slotNum：由使用方指定 slot 的个数，默认为 10
• interval：由使用方指定每个 slot 对应的时间范围，默认为 1 秒

初始化时间轮实例的过程中，会完成定时器 ticker 以及各个 channel 的初始化，并针对数组 中的各个 slot 进行初始化，每个 slot 位置都需要填充一个 list.

每个时间轮实例都会异步调用 run 方法，启动一个常驻 goroutine 用于接收和处理定时任务.

// 创建单机版时间轮 slotNum——时间轮环状数组长度  interval——扫描时间间隔
func NewTimeWheel(slotNum int, interval time.Duration) *TimeWheel {// 环状数组长度默认为 10if slotNum <= 0 {slotNum = 10}// 扫描时间间隔默认为 1 秒if interval <= 0 {interval = time.Second}// 初始化时间轮实例t := TimeWheel{interval:     interval,ticker:       time.NewTicker(interval),stopc:        make(chan struct{}),keyToETask:   make(map[string]*list.Element),slots:        make([]*list.List, 0, slotNum),addTaskCh:    make(chan *taskElement),removeTaskCh: make(chan string),}for i := 0; i < slotNum; i++ {t.slots = append(t.slots, list.New())}// 异步启动时间轮常驻 goroutinego t.run()return &t
}

构造函数比较简单，由于异步run启动时间轮常驻 goroutine，所以我们现在看看run方法。

启动

时间轮运行的核心逻辑位于 timeWheel.run 方法中，该方法会通过 for 循环结合 select 多路复用的方式运行，属于 golang 中非常常见的异步编程风格.

goroutine 运行过程中需要从以下四类 channel 中接收不同的信号，并进行逻辑的分发处理：

• stopc：停止时间轮，使得当前 goroutine 退出
• ticker：接收到 ticker 的信号说明时间由往前推进了一个 interval，则需要批量检索并执行当前 slot 中的定时任务. 并推进指针 curSlot 往前偏移
• addTaskCh：接收创建定时任务的指令
• removeTaskCh：接收删除定时任务的指令

此处值得一提的是，后续不论是创建、删除还是检索定时任务，都是通过这个常驻 goroutine 完成的，因此在访问一些临界资源的时候，不需要加锁，因为不存在并发访问的情况

// 运行时间轮
func (t *TimeWheel) run() {defer func() {if err := recover(); err != nil {// ...}}()// 通过 for + select 的代码结构运行一个常驻 goroutine 是常规操作for {select {// 停止时间轮case <-t.stopc:return// 接收到定时信号case <-t.ticker.C:// 批量执行定时任务t.tick()// 接收创建定时任务的信号case task := <-t.addTaskCh:t.addTask(task)// 接收到删除定时任务的信号case removeKey := <-t.removeTaskCh:t.removeTask(removeKey)}}
}

停止

时间轮提供了一个 Stop 方法，用于手动停止时间轮，回收对应的 goroutine 和 ticker 资源.

停止时间轮的操作是通过关闭 stopc channel 完成的，由于 channel 不允许被反复关闭，因此这里通过 sync.Once 保证该逻辑只被调用一次.

// 停止时间轮
func (t *TimeWheel) Stop() {// 通过单例工具，保证 channel 只能被关闭一次，避免 panict.Do(func() {// 定制定时器 tickert.ticker.Stop()// 关闭定时器运行的 stopcclose(t.stopc)})
}

创建任务

创建一笔定时任务的核心步骤如下：

• 使用方往 addTaskCh 中投递定时任务，由常驻 goroutine 接收定时任务
• 根据执行时间，推算出定时任务所处的 slot 位置以及需要延迟的轮次 cycle
• 将定时任务包装成一个 list node，追加到对应 slot 位置的 list 尾部
• 以定时任务唯一键为 key，list node 为 value，在 keyToETask map 中建立映射关系，方便后续删除任务时使用

我们首先看一下源码，然后再看相应的图解。

AddTask

// 添加定时任务到时间轮中
func (t *TimeWheel) AddTask(key string, task func(), executeAt time.Time) {// 根据执行时间推算得到定时任务从属的 slot 位置，以及需要延迟的轮次pos, cycle := t.getPosAndCircle(executeAt)// 将定时任务通过 channel 进行投递t.addTaskCh <- &taskElement{pos:   pos,cycle: cycle,task:  task,key:   key,}
}

pos, cycle := t.getPosAndCircle(executeAt): 这行代码调用了TimeWheel的另一个方法getPosAndCircle，传入期望执行的时间executeAt。这个方法计算出任务应该放置在时间轮的哪个槽位上（pos），以及在任务第一次执行前，时间轮需要转过多少完整的圈数（cycle）。

t.addTaskCh <- &taskElement{: 这是Go语言的通道（channel）操作。它创建了一个taskElement结构体实例，并通过TimeWheel中的addTaskCh通道发送出去。这种方式通常用于跨goroutine的安全通信，意味着AddTask方法将定时任务提交到另一个可能在不同goroutine中运行的执行上下文。

• pos: pos,: 设置taskElement的pos字段，表示这个任务在时间轮的哪一个位置。
• cycle: cycle,: 设置taskElement的cycle字段，表示任务在能被执行前时间轮需要转动多少圈。
• task: task,: 将外部传入的任务闭包task赋给taskElement。
• key: key,: 将任务的唯一标识符key赋给taskElement。

getPosAndCircle

// 根据执行时间推算得到定时任务从属的 slot 位置，以及需要延迟的轮次
func (t *TimeWheel) getPosAndCircle(executeAt time.Time) (int, int) {delay := int(time.Until(executeAt))// 定时任务的延迟轮次cycle := delay / (len(t.slots) * int(t.interval))// 定时任务从属的环状数组 indexpos := (t.curSlot + delay/int(t.interval)) % len(t.slots)return pos, cycle
}

为了举例说明这个函数如何工作，我们需要设定一些参数：

• 假设时间轮TimeWheel的slots有60个槽位，代表一分钟内的每一秒（len(t.slots) = 60）。
• 时间轮的每个槽位对应1秒钟（t.interval = 1秒）。
• 假设当前时间轮的指针curSlot在第0槽位上（t.curSlot = 0），这通常表示整点时刻。
• 设定一个将来的时间点executeAt，假设这个时间点是从现在开始的第62秒后。这意味着我们希望在1分钟2秒后执行任务（delay = 62秒）。

// 从现在开始到执行时间的延迟时间（秒）
delay := int(time.Until(executeAt))  // delay = 62// 计算定时任务需要经过的完整时间轮循环数
cycle := delay / (len(t.slots) * int(t.interval))  
// cycle = 62 / (60 * 1) = 1.033，向下取整为 1// 计算定时任务应该位于的槽位（数组index）
pos := (t.curSlot + delay/int(t.interval)) % len(t.slots)
// pos = (0 + 62/1) % 60 = 62 % 60 = 2

所以，函数getPosAndCircle将会返回(2, 1)：

假设时间轮有5个槽位，每个槽位间隔为1秒，并且当前槽位（curSlot）为0。我们需要计算延迟0到11秒的任务对应的槽位（pos）和轮次（cycle）。

• 延迟0秒：槽位0，轮次0
• 延迟1秒：槽位1，轮次0
• 延迟2秒：槽位2，轮次0
• 延迟3秒：槽位3，轮次0
• 延迟4秒：槽位4，轮次0
• 延迟5秒：槽位0，轮次1
• 延迟6秒：槽位1，轮次1
• 延迟7秒：槽位2，轮次1
• 延迟8秒：槽位3，轮次1
• 延迟9秒：槽位4，轮次1
• 延迟10秒：槽位0，轮次2
• 延迟11秒：槽位1，轮次2

现在看一下执行过程。

addTask

// 常驻 goroutine 接收到创建定时任务后的处理逻辑
func (t *TimeWheel) addTask(task *taskElement) {// 获取到定时任务从属的环状数组 index 以及对应的 listlist := t.slots[task.pos]// 倘若定时任务 key 之前已存在，则需要先删除定时任务if _, ok := t.keyToETask[task.key]; ok {t.removeTask(task.key)}// 将定时任务追加到 list 尾部eTask := list.PushBack(task)// 建立定时任务 key 到将定时任务所处的节点t.keyToETask[task.key] = eTask
}

倘若定时任务 key 之前已存在，则需要先删除定时任务，然后重新添加到末尾。这张图很详细的说明执行的过程了。

删除任务

删除一笔定时任务的核心步骤如下：

• 使用方往 removeTaskCh 中投递删除任务的 key，由常驻 goroutine 接收处理
• 从 keyToETask map 中，找到该任务对应的 list node
• 从 keyToETask map 中移除该组 kv 对
• 从对应 slot 的 list 中移除该 list node

// 删除定时任务，投递信号
func (t *TimeWheel) RemoveTask(key string) {t.removeTaskCh <- key
}
// 时间轮常驻 goroutine 接收到删除任务信号后，执行的删除任务逻辑
func (t *TimeWheel) removeTask(key string) {eTask, ok := t.keyToETask[key]if !ok {return}// 将定时任务节点从映射 map 中移除delete(t.keyToETask, key)// 获取到定时任务节点后，将其从 list 中移除task, _ := eTask.Value.(*taskElement)_ = t.slots[task.pos].Remove(eTask)
}

执行定时任务

最后来捋一下最核心的链路——检索并批量执行定时任务的流程.

首先，每当接收到 ticker 信号时，会根据当前的 curSlot 指针，获取到对应 slot 位置挂载的定时任务 list，调用 execute 方法执行其中的定时任务，最后通过 circularIncr 方法推进 curSlot 指针向前移动。

// 常驻 goroutine 每次接收到定时信号后用于执行定时任务的逻辑
func (t *TimeWheel) tick() {// 根据 curSlot 获取到当前所处的环状数组索引位置，取出对应的 listlist := t.slots[t.curSlot]// 在方法返回前，推进 curSlot 指针的位置，进行环状遍历defer t.circularIncr()// 批量处理满足执行条件的定时任务t.execute(list)
}

在 execute 方法中，会对 list 中的定时任务进行遍历：

• 对于 cycle > 0 的定时任务，说明当前还未达到执行条件，需要将其 cycle 值减 1，留待后续轮次再处理
• 对于 cycle = 0 的定时任务，开启一个 goroutine ，执行其中的闭包函数 task，并将其从 list 和 map 中移除

// 执行定时任务，每次处理一个 list
func (t *TimeWheel) execute(l *list.List) {// 遍历 listfor e := l.Front(); e != nil; {// 获取到每个节点对应的定时任务信息taskElement, _ := e.Value.(*taskElement)// 倘若任务还存在延迟轮次，则只对 cycle 计数器进行扣减，本轮不作任务的执行if taskElement.cycle > 0 {taskElement.cycle--e = e.Next()continue}// 当前节点对应定时任务已达成执行条件，开启一个 goroutine 负责执行任务go func() {defer func() {if err := recover(); err != nil {// ...}}()taskElement.task()}()// 任务已执行，需要把对应的任务节点从 list 中删除next := e.Next()l.Remove(e)// 把任务 key 从映射 map 中删除delete(t.keyToETask, taskElement.key)e = next}
}// 每次 tick 后需要推进 curSlot 指针的位置，slots 在逻辑意义上是环状数组，所以在到达尾部时需要从新回到头部 
func (t *TimeWheel) circularIncr() {t.curSlot = (t.curSlot + 1) % len(t.slots)
}

---

总结

看了小徐先生的推文跟B站视频收获很多，也期待后续跟着大佬继续学习。

参考

https://zhuanlan.zhihu.com/p/658079556
https://blog.csdn.net/YouMing_Li/article/details/134089794