設(shè)計思路

在linux下實現(xiàn)定時器主要有如下方式

• 基于鏈表實現(xiàn)定時器
• 基于排序鏈表實現(xiàn)定時器
• 基于最小堆實現(xiàn)定時器
• 基于時間輪實現(xiàn)定時器

在這當中基于時間輪方式實現(xiàn)的定時器時間復雜度最小，效率最高，然而我們可以通過優(yōu)先隊列實現(xiàn)時間輪定時器。

優(yōu)先隊列的實現(xiàn)可以使用最大堆和最小堆，因此在隊列中所有的數(shù)據(jù)都可以定義排序規(guī)則自動排序。我們直接通過隊列中pop函數(shù)獲取數(shù)據(jù)，就是我們按照自定義排序規(guī)則想要的數(shù)據(jù)。

在Golang中實現(xiàn)一個優(yōu)先隊列異常簡單，在container/head包中已經(jīng)幫我們封裝了，實現(xiàn)的細節(jié)，我們只需要實現(xiàn)特定的接口就可以。

下面是官方提供的例子

// 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
}

因為優(yōu)先隊列底層數(shù)據(jù)結(jié)構(gòu)是由二叉樹構(gòu)建的，所以我們可以通過數(shù)組來保存二叉樹上的每一個節(jié)點。
改數(shù)組需要實現(xiàn)Go預先定義的接口Len,Less,Swap,Push,Pop和update。

• Len 接口定義返回隊列長度
• Swap 接口定義隊列數(shù)據(jù)優(yōu)先級，比較規(guī)則
• Push 接口定義push數(shù)據(jù)到隊列中操作
• Pop 接口定義返回隊列中頂層數(shù)據(jù)，并且將改數(shù)據(jù)刪除
• update 接口定義更新隊列中數(shù)據(jù)信息

接下來我們分析 https://github.com/leesper/tao 開源的代碼中TimeingWheel 中的實現(xiàn)細節(jié)。

一、設(shè)計細節(jié)

1. 結(jié)構(gòu)細節(jié)

1.1 定時任務(wù)結(jié)構(gòu)

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結(jié)構(gòu)是定時任務(wù)抽象結(jié)構(gòu)

• id 定時任務(wù)的唯一id，可以這個id查找在隊列中的定時任務(wù)
• expiration 定時任務(wù)的到期時間點，當?shù)竭@個時間點后，觸發(fā)定時任務(wù)的執(zhí)行，在優(yōu)先隊列中也是通過這個字段來排序
• interval 定時任務(wù)的觸發(fā)頻率，每隔interval時間段觸發(fā)一次
• timeout 這個結(jié)構(gòu)中保存定時超時任務(wù)，這個任務(wù)函數(shù)參數(shù)必須符合相應(yīng)的接口類型
• index 保存在隊列中的任務(wù)所在的下標

1.2 時間輪結(jié)構(gòu)

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來保存，已經(jīng)觸發(fā)的定時任務(wù)
• timers 是[]*timerType類型的slice，保存所有定時任務(wù)
• ticker 當每一個ticker到來時，時間輪都會檢查隊列中head元素是否到達超時時間
• wg 用于并發(fā)控制
• addChan 通過帶緩存的chan來向隊列中添加任務(wù)
• cancelChan 定時器停止的chan
• sizeChan 返回隊列中任務(wù)的數(shù)量的chan
• ctx 和 cancel 用戶并發(fā)控制

2. 關(guān)鍵函數(shù)實現(xiàn)

2.1 TimingWheel的主循環(huán)函數(shù)

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函數(shù)，當創(chuàng)建一個TimeingWheel時，通過一個goroutine來執(zhí)行start,在start中for循環(huán)和select來監(jiān)控不同的channel的狀態(tài)

• <-tw.cancelChan 返回要取消的定時任務(wù)的id，并且在隊列中刪除
• tw.sizeChan <- 將定時任務(wù)的個數(shù)放入這個無緩存的channel中
• <-tw.ctx.Done() 當父context執(zhí)行cancel時，該channel 就會有數(shù)值，表示該TimeingWheel 要停止
• <-tw.addChan 通過帶緩存的addChan來向隊列中添加任務(wù)
• <-tw.ticker.C ticker定時，當每一個ticker到來時，time包就會向該channel中放入當前Time，當每一個Ticker到來時，TimeingWheel都需要檢查隊列中到到期的任務(wù)（tw.getExpired()），通過range來放入TimeOutChannelchannel中， 最后在更新隊列。

2.2 TimingWheel的尋找超時任務(wù)函數(shù)

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循環(huán)從隊列中取數(shù)據(jù)，直到該隊列為空或者是遇見第一個當前時間比任務(wù)開始時間大的任務(wù)，append到expired中。因為優(yōu)先隊列中是根據(jù)expiration來排序的，
所以當取到第一個定時任務(wù)未到的任務(wù)時，表示該定時任務(wù)以后的任務(wù)都未到時間。

2.3 TimingWheel的更新隊列函數(shù)

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函數(shù)取出隊列中要執(zhí)行的任務(wù)時，當有的定時任務(wù)需要不斷執(zhí)行，所以就需要判斷是否該定時任務(wù)需要重新放回優(yōu)先隊列中。isRepeat是通過判斷任務(wù)中interval是否大于 0 判斷，
如果大于0 則，表示永久就生效。

3. TimeingWheel 的用法

防止外部濫用，阻塞定時器協(xié)程，框架又一次封裝了timer這個包，名為timer_wapper這個包,它提供了兩種調(diào)用方式。

3.1 第一種普通的調(diào)用定時任務(wù)

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 添加定時器任務(wù)，任務(wù)在定時器協(xié)程執(zhí)行
• when為執(zhí)行時間
• interv為執(zhí)行周期，interv=0只執(zhí)行一次
• cb為回調(diào)函數(shù)

3.2 第二種通過任務(wù)池調(diào)用定時任務(wù)

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)
        }))
}

參數(shù)和上面的參數(shù)一樣，只是在第三個參數(shù)中使用了任務(wù)池，將定時任務(wù)放入了任務(wù)池中。定時任務(wù)的本身執(zhí)行就是一個put操作。
至于put以后，那就是workers這個包管理的了。在worker包中， 也就是維護了一個任務(wù)池，任務(wù)池中的任務(wù)會有序的執(zhí)行，方便管理。