设计思路
在linux下实现定时器主要有如下方式
- 基于链表实现定时器
- 基于排序链表实现定时器
- 基于最小堆实现定时器
- 基于时间轮实现定时器
在这当中基于时间轮方式实现的定时器时间复杂度最小，效率最高，然而我们可以通过优先队列实现时间轮定时器。
优先队列的实现可以使用最大堆和最小堆，因此在队列中所有的数据都可以定义排序规则自动排序。我们直接通过队列中pop函数获取数据，就是我们按照自定义排序规则想要的数据。
在Golang中实现一个优先队列异常简单，在container/head包中已经帮我们封装了，实现的细节，我们只需要实现特定的接口就可以。
下面是官方提供的例子
// This example demonstrates a priority queue built using the heap interface.
// An Item is something we manage in a priority queue.
type Item struct {
value string // The value of the item; arbitrary.
priority int // The priority of the item in the queue.
// The index is needed by update and is maintained by the heap.Interface methods.
index int // The index of the item in the heap.
}
// A PriorityQueue implements heap.Interface and holds Items.
type PriorityQueue []*Item
func (pq PriorityQueue) Len() int { return len(pq) }
func (pq PriorityQueue) Less(i, j int) bool {
// We want Pop to give us the highest, not lowest, priority so we use greater than here.
return pq[i].priority > pq[j].priority
}
func (pq PriorityQueue) Swap(i, j int) {
pq[i], pq[j] = pq[j], pq[i]
pq[i].index = i
pq[j].index = j
}
func (pq *PriorityQueue) Push(x interface{}) {
n := len(*pq)
item := x.(*Item)
item.index = n
*pq = append(*pq, item)
}
func (pq *PriorityQueue) Pop() interface{} {
old := *pq
n := len(old)
item := old[n-1]
item.index = -1 // for safety
*pq = old[0 : n-1]
return item
}
因为优先队列底层数据结构是由二叉树构建的，所以我们可以通过数组来保存二叉树上的每一个节点。
改数组需要实现Go预先定义的接口Len,Less,Swap,Push,Pop和update。
-
Len接口定义返回队列长度 -
Swap接口定义队列数据优先级，比较规则 -
Push接口定义push数据到队列中操作 -
Pop接口定义返回队列中顶层数据，并且将改数据删除 -
update接口定义更新队列中数据信息
接下来我们分析 https://github.com/leesper/tao 开源的代码中TimeingWheel 中的实现细节。
一、设计细节
1. 结构细节
1.1 定时任务结构
type timerType struct {
id int64
expiration time.Time
interval time.Duration
timeout *OnTimeOut
index int // for container/heap
}
type OnTimeOut struct {
Callback func(time.Time, WriteCloser)
Ctx context.Context
}
timerType结构是定时任务抽象结构
-
id定时任务的唯一id，可以这个id查找在队列中的定时任务 -
expiration定时任务的到期时间点，当到这个时间点后，触发定时任务的执行，在优先队列中也是通过这个字段来排序 -
interval定时任务的触发频率，每隔interval时间段触发一次 -
timeout这个结构中保存定时超时任务，这个任务函数参数必须符合相应的接口类型 -
index保存在队列中的任务所在的下标
1.2 时间轮结构
type TimingWheel struct {
timeOutChan chan *OnTimeOut
timers timerHeapType
ticker *time.Ticker
wg *sync.WaitGroup
addChan chan *timerType // add timer in loop
cancelChan chan int64 // cancel timer in loop
sizeChan chan int // get size in loop
ctx context.Context
cancel context.CancelFunc
}
-
timeOutChan定义一个带缓存的chan来保存，已经触发的定时任务 -
timers是[]*timerType类型的slice，保存所有定时任务 -
ticker当每一个ticker到来时，时间轮都会检查队列中head元素是否到达超时时间 -
wg用于并发控制 -
addChan通过带缓存的chan来向队列中添加任务 -
cancelChan定时器停止的chan -
sizeChan返回队列中任务的数量的chan -
ctx和cancel用户并发控制
2. 关键函数实现
2.1 TimingWheel的主循环函数
func (tw *TimingWheel) start() {
for {
select {
case timerID := <-tw.cancelChan:
index := tw.timers.getIndexByID(timerID)
if index >= 0 {
heap.Remove(&tw.timers, index)
}
case tw.sizeChan <- tw.timers.Len():
case <-tw.ctx.Done():
tw.ticker.Stop()
return
case timer := <-tw.addChan:
heap.Push(&tw.timers, timer)
case <-tw.ticker.C:
timers := tw.getExpired()
for _, t := range timers {
tw.TimeOutChannel() <- t.timeout
}
tw.update(timers)
}
}
}
首先的start函数，当创建一个TimeingWheel时，通过一个goroutine来执行start,在start中for循环和select来监控不同的channel的状态
-
<-tw.cancelChan返回要取消的定时任务的id，并且在队列中删除 -
tw.sizeChan <-将定时任务的个数放入这个无缓存的channel中 -
<-tw.ctx.Done()当父context执行cancel时，该channel 就会有数值，表示该TimeingWheel要停止 -
<-tw.addChan通过带缓存的addChan来向队列中添加任务 -
<-tw.ticker.Cticker定时，当每一个ticker到来时，time包就会向该channel中放入当前Time，当每一个Ticker到来时，TimeingWheel都需要检查队列中到到期的任务（tw.getExpired()），通过range来放入TimeOutChannelchannel中， 最后在更新队列。
2.2 TimingWheel的寻找超时任务函数
func (tw *TimingWheel) getExpired() []*timerType {
expired := make([]*timerType, 0)
for tw.timers.Len() > 0 {
timer := heap.Pop(&tw.timers).(*timerType)
elapsed := time.Since(timer.expiration).Seconds()
if elapsed > 1.0 {
dylog.Warn(0, "timing_wheel", nil, "elapsed %d", elapsed)
}
if elapsed > 0.0 {
expired = append(expired, timer)
continue
} else {
heap.Push(&tw.timers, timer)
break
}
}
return expired
}
通过for循环从队列中取数据，直到该队列为空或者是遇见第一个当前时间比任务开始时间大的任务，append到expired中。因为优先队列中是根据expiration来排序的，
所以当取到第一个定时任务未到的任务时，表示该定时任务以后的任务都未到时间。
2.3 TimingWheel的更新队列函数
func (tw *TimingWheel) update(timers []*timerType) {
if timers != nil {
for _, t := range timers {
if t.isRepeat() { // repeatable timer task
t.expiration = t.expiration.Add(t.interval)
// if task time out for at least 10 seconds, the expiration time needs
// to be updated in case this task executes every time timer wakes up.
if time.Since(t.expiration).Seconds() >= 10.0 {
t.expiration = time.Now()
}
heap.Push(&tw.timers, t)
}
}
}
}
当getExpired函数取出队列中要执行的任务时，当有的定时任务需要不断执行，所以就需要判断是否该定时任务需要重新放回优先队列中。isRepeat是通过判断任务中interval是否大于 0 判断，
如果大于0 则，表示永久就生效。
3. TimeingWheel 的用法
防止外部滥用，阻塞定时器协程，框架又一次封装了timer这个包，名为timer_wapper这个包,它提供了两种调用方式。
3.1 第一种普通的调用定时任务
func (t *TimerWrapper) AddTimer(when time.Time, interv time.Duration, cb TimerCallback) int64{
return t.TimingWheel.AddTimer(
when,
interv,
serverbase.NewOnTimeOut(t.ctx, func(t time.Time, c serverbase.WriteCloser) {
cb()
}))
}
- AddTimer 添加定时器任务，任务在定时器协程执行
- when为执行时间
- interv为执行周期，interv=0只执行一次
- cb为回调函数
3.2 第二种通过任务池调用定时任务
func (t *TimerWrapper) AddTimerInPool(when time.Time, interv time.Duration, cb TimerCallback) int64 {
return t.TimingWheel.AddTimer(
when,
interv,
serverbase.NewOnTimeOut(t.ctx, func(t time.Time, c serverbase.WriteCloser) {
workpool.WorkerPoolInstance().Put(cb)
}))
}
参数和上面的参数一样，只是在第三个参数中使用了任务池，将定时任务放入了任务池中。定时任务的本身执行就是一个put操作。
至于put以后，那就是workers这个包管理的了。在worker包中， 也就是维护了一个任务池，任务池中的任务会有序的执行，方便管理。
有疑问加站长微信联系（非本文作者）
