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

Go实现单机版时间轮

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

https://github.com/xiaoxuxiansheng/timewheel

package main

import (
	"container/list"
	"fmt"
	"sync"
	"time"
)

type taskElement struct {
	task  func()
	pos   int
	cycle int
	key   string
}

type TimeWheel struct {
	sync.Once
	interval     time.Duration
	ticker       *time.Ticker
	stopc        chan struct{}
	addTaskCh    chan *taskElement
	removeTaskCh chan string
	slots        []*list.List
	curSlot      int
	keyToETask   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:
			return
		case <-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) {
	// 遍历每个 list
	for 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.Once
	interval     time.Duration
	ticker       *time.Ticker
	stopc        chan struct{}
	addTaskCh    chan *taskElement
	removeTaskCh chan string
	slots        []*list.List
	curSlot      int
	keyToETask   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 {
    // 环状数组长度默认为 10
    if 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())
    }
    
    // 异步启动时间轮常驻 goroutine
    go 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 只能被关闭一次，避免 panic
    t.Do(func() {
        // 定制定时器 ticker
        t.ticker.Stop()
        // 关闭定时器运行的 stopc
        close(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))
    // 定时任务从属的环状数组 index
    pos := (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 以及对应的 list
    list := 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 获取到当前所处的环状数组索引位置，取出对应的 list
    list := 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) {
    // 遍历 list
    for 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)
}