手动实现一个时间轮
// 包 timewheel 提供了一个简单的时间轮调度器的实现。
package timewheel
import (
"container/list" // 引入 list 包来使用双向链表
"context" // 引入 context 包来进行上下文控制
"time" // 引入 time 包来处理时间相关的功能
)
// Task 接口定义了需要时间轮执行的任务，它必须实现 Execute 方法。
type Task interface {
Execute()
}
// TimeNode 结构体表示时间轮上的一个节点。
type TimeNode struct {
task Task // 任务实例
excTime time.Time // 预定的执行时间
cycle int // 时间轮需要转动多少圈后执行任务
}
// TimeWheel 结构体表示时间轮。
type TimeWheel struct {
startTime time.Time // 时间轮的启动时间
interval time.Duration // 时间轮的时间间隔
currentPos int // 当前的槽位置
ticker *time.Ticker // 定时触发器
slots []*list.List // 时间轮的槽数组，每个槽点是一个链表
tasks chan Task // 任务channel，用于执行任务
stop chan struct{} // 停止channel，用于停止时间轮
cancel context.CancelFunc // 上下文取消函数
}
// NewTimeWheel 创建并初始化一个时间轮实例。
func NewTimeWheel(interval time.Duration, slotNum int) *TimeWheel {
slots := make([]*list.List, slotNum)
for i := range slots {
slots[i] = list.New()
}
return &TimeWheel{
startTime: time.Now().UTC(),
interval: interval,
currentPos: 0,
ticker: time.NewTicker(interval),
slots: slots,
tasks: make(chan Task, 1000),
stop: make(chan struct{}),
}
}
// DefaultTimeWheel 创建一个默认配置的时间轮实例。
func DefaultTimeWheel() *TimeWheel {
return NewTimeWheel(time.Second, 60)
}
// Start 启动时间轮，使其开始调度任务。
func (t *TimeWheel) Start() {
ctx := context.Background()
ctx, cancelFunc := context.WithCancel(ctx)
t.cancel = cancelFunc
go t.Loop(ctx)
go func() {
for {
select {
case task := <-t.tasks: // 当任务channel中有任务时，执行任务
task.Execute()
case <-t.stop: // 当接收到停止信号时，停止定时器并退出
t.ticker.Stop()
return
}
}
}()
}
// Stop 停止时间轮，取消所有调度。
func (t *TimeWheel) Stop() {
t.stop <- struct{}{} // 发送停止信号
t.cancel() // 调用上下文的取消函数
}
// Loop 是时间轮的核心调度循环。
func (t *TimeWheel) Loop(ctx context.Context) {
for {
select {
case <-ctx.Done(): // 当上下文被取消时退出循环
return
case <-t.ticker.C: // 定时器触发时执行
t.currentPos = (t.currentPos + 1) % len(t.slots) // 计算当前槽位置
l := t.slots[t.currentPos]
for e := l.Front(); e != nil; {
node := e.Value.(*TimeNode)
if node.cycle > 0 {
node.cycle-- // 减少圈数
e = e.Next()
continue
}
t.tasks <- node.task // 将任务发送到任务channel
next := e.Next()
l.Remove(e) // 移除已执行的任务
e = next
}
}
}
}
// AddTask 添加任务到时间轮。
func (t *TimeWheel) AddTask(task Task, delay time.Duration) {
pos := (t.currentPos + int(delay/t.interval)) % len(t.slots) // 计算任务应该放入的槽位置
cycle := int(delay/t.interval) / len(t.slots) // 计算任务需要等待的圈数
if t.slots[pos] == nil {
t.slots[pos] = list.New()
}
t.slots[pos].PushBack(
&TimeNode{
task: task,
excTime: time.Now().Add(delay),
cycle: cycle,
},
)
}
说明文档
1. 概述
时间轮（TimeWheel）是一个用于任务调度的数据结构，它允许你以固定的时间间隔调度任务。这段Go代码提供了一个简单的时间轮实现。
2. 结构体和接口
-
Task: 一个接口，所有希望被时间轮调度的任务都应该实现这个接口的Execute方法。 -
TimeNode: 代表时间轮上的节点，持有一个任务和任务的执行时间以及剩余圈数。 -
TimeWheel: 时间轮的主体，包含了时间轮的各项配置，如时间间隔、槽数组等，并且可以开始和停止任务调度。
3. 方法
-
NewTimeWheel: 创建并返回一个新的时间轮实例。 -
DefaultTimeWheel: 创建一个拥有默认配置（一秒钟的间隔，60个槽）的时间轮实例。 -
Start: 启动时间轮，开始调度任务。 -
Stop: 停止时间轮，取消所有调度任务。 -
Loop: 作为时间轮的核心调度循环，负责移动槽位置，执行到期的任务。 -
AddTask: 向时间轮添加任务，可以指定延迟时间。
4. 使用方式
创建时间轮实例，通过 AddTask 方法添加任务，然后调用 Start 方法开始调度。当不再需要时间轮时，可以调用 Stop 方法来停止所有调度。
5. 注意事项
- 时间轮初始化时会创建一个任务通道
tasks，其大小为1000，因此它可以在不阻塞的情况下缓存多达1000个待执行的任务。 - 时间轮的
Loop方法使用了Go的select语句来高效的处理任务的执行和时间轮的停止。 - 在停止时间轮时，应确保所有的任务都已经完成或不需要执行，以避免资源泄露或未完成的任务。
