北京网站开发联系电话,wordpress 5.0.2 发布失败,长春企业网站如何建设,erp项目管理软件Go实现单机版时间轮
上一章介绍了时间轮的相关概念#xff0c;接下来我们会使用 golang 标准库的定时器工具 time ticker 结合环状数组的设计思路#xff0c;实现一个单机版的单级时间轮。 首先我们先运行一下下面的源码#xff0c;看一下如何使用。
https://github.com/x…Go实现单机版时间轮
上一章介绍了时间轮的相关概念接下来我们会使用 golang 标准库的定时器工具 time ticker 结合环状数组的设计思路实现一个单机版的单级时间轮。 首先我们先运行一下下面的源码看一下如何使用。
https://github.com/xiaoxuxiansheng/timewheel
package mainimport (container/listfmtsynctime
)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 m0.000173292
test1, 2023-11-03 13:02:58.043555 0800 CST m1.000891376
test2, 2023-11-03 13:03:00.043567 0800 CST m3.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用于停止 goroutineaddTaskCh用于接收创建定时器指令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 的个数默认为 10interval由使用方指定每个 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 尾部以定时任务唯一键为 keylist 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
