Golang-基于TimeingWheel定时器

wiseAaron · · 687 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

设计思路

在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,Popupdate

  • 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
  • ctxcancel 用户并发控制

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.C ticker定时,当每一个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循环从队列中取数据,直到该队列为空或者是遇见第一个当前时间比任务开始时间大的任务,appendexpired中。因为优先队列中是根据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包中, 也就是维护了一个任务池,任务池中的任务会有序的执行,方便管理。


有疑问加站长微信联系(非本文作者)

本文来自:简书

感谢作者:wiseAaron

查看原文:Golang-基于TimeingWheel定时器

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:701969077

687 次点击  
加入收藏 微博
被以下专栏收入,发现更多相似内容
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传