用Golang實作基于時間輪算法的定時器

關于什麼是時間輪算法，自己百度吧，這裡列出我用go語言實作的時間輪算法，已經上線應用，穩定。

package timer

import (
    "log"
    "sync"
    "time"
)

const wheel_cnt uint8 =                                                                   //時間輪數量5個
var element_cnt_per_wheel = [wheel_cnt]uint32,,,,}                          //每個時間輪的槽(元素)數量。在 256+64+64+64+64 = 512 個槽中，表示的範圍為 2^32
var right_shift_per_wheel = [wheel_cnt]uint32,,,,}                                //當指針指向目前時間輪最後一位數，再走一位就需要向上進位。每個時間輪進位的時候，使用右移的方式，最快實作進位。這裡是每個輪的進位二進制位數
var base_per_wheel = [wheel_cnt]uint32,, *, * *, * * *} //記錄每個時間輪指針目前指向的位置
var mutex sync.Mutex                                                                        //加鎖
var rwmutex sync.RWMutex
var newest [wheel_cnt]uint32                           //每個時間輪目前指針所指向的位置
var timewheels ][]*Node                              //定義5個時間輪
var TimerMap map[string]*Node = make(map[string]*Node) //儲存待執行的計時器，友善按連結清單節點指針位址直接删除定時器

type Timer struct {
    Name        string            //定時器名稱
    Inteval     uint32            //時間間隔，即以插入該定時器的時間為起點，Inteval秒之後執行回調函數DoSomething()。例如程序插入該定時器的時間是2015-04-05 10:23:00，Inteval=5，則執行DoSomething()的時間就是2015-04-05 10:23:05。
    DoSomething func(interface{}) //自定義事件處理函數，需要觸發的事件
    Args        interface{}       //上述函數的輸入參數
}

func SetTimer(name string, inteval uint32, handler func(interface{}), args interface{}) {
    if inteval <= {
        return
    }
    var bucket_no uint8 =
    var offset uint32 = inteval
    var left uint32 = inteval
    for offset >= element_cnt_per_wheel[bucket_no] { //偏移量大于目前時間輪容量，則需要向高位進位
        offset >>= right_shift_per_wheel[bucket_no] //計算高位的值。偏移量除以低位的進制。比如低位目前是256，則右移8個二進制位，就是除以256，得到的結果是高位的值。
        var tmp uint32 =
        if bucket_no == {
            tmp =
        }
        left -= base_per_wheel[bucket_no] * (element_cnt_per_wheel[bucket_no] - newest[bucket_no] - tmp)
        bucket_no++
    }
    if offset < {
        return
    }
    if inteval < base_per_wheel[bucket_no]*offset {
        return
    }
    left -= base_per_wheel[bucket_no] * (offset -)
    pos := (newest[bucket_no] + offset) % element_cnt_per_wheel[bucket_no] //通過類似hash的方式，找到在時間輪上的插入位置

    var node Node
    node.SetData(Timer{name, left, handler, args})

    rwmutex.RLock()
    TimerMap[name] = timewheels[bucket_no][pos].InsertHead(node) //插入定時器
    rwmutex.RUnlock()
    //fmt.Println("pos ", bucket_no, pos, tmp)
}

func step() {
    //var dolist list.List
    {
        rwmutex.RLock()
        //周遊所有桶
        var bucket_no uint8 =
        for bucket_no =; bucket_no < wheel_cnt; bucket_no++ {
            newest[bucket_no] = (newest[bucket_no] +) % element_cnt_per_wheel[bucket_no] //目前指針遞增1
            //fmt.Println(newest)
            var head *Node = timewheels[bucket_no][newest[bucket_no]] //傳回目前指針指向的槽位置的表頭
            var firstElement *Node = head.Next()
            for firstElement != nil { //連結清單不為空
                if value, ok := firstElement.Data().(Timer); ok { //如果element裡面确實存儲了Timer類型的數值，那麼ok傳回true，否則傳回false。
                    inteval := value.Inteval
                    doSomething := value.DoSomething
                    args := value.Args
                    if nil != doSomething { //有遇到函數為nil的情況，是以這裡判斷下非nil
                        if == bucket_no || == inteval {
                            //dolist.PushBack(value) //執行自定義處理函數
                            go doSomething(args)
                        } else {
                            SetTimer(value.Name, inteval, doSomething, args) //重新插入計時器
                        }
                    }
                    Delete(firstElement) //删除定時器
                }
                firstElement = head.Next() //重新定位到連結清單第一個元素頭
            }
            if != newest[bucket_no] { //指針不是0，還未轉回到原點，跳出。如果回到原點，則說明轉完了一圈，需要向高位進位1，則繼續循環入高位步進一步。
                break
            }
        }
        rwmutex.RUnlock()
    }
}

func Run() {
    var i int =
    for {
        go step()
        i++
        log.Printf("第%ds", i)
        //間隔時間inteval=1s
        time.Sleep * time.Second)
    }
}

func init() { //初始化
    var bucket_no uint8 =
    for bucket_no =; bucket_no < wheel_cnt; bucket_no++ {
        var i uint32 =
        for ; i < element_cnt_per_wheel[bucket_no]; i++ {
            timewheels[bucket_no] = append(timewheels[bucket_no], new(Node))
        }
    }
}
           

測試下效果

package main

import (
    "log"
    "runtime"
    "timer_server/timer"
)

func callback1(args interface{}) {
    //隻執行一次的事件
    if values, ok := args.([]string); ok {
        var str1 string = values]
        var str2 string = values]
        log.Println("callback1(" + str1 + "," + str2 + ")")
    } else {
        log.Println("callback1()")
    }
}

func callback2(args interface{}) {
    //每次在目前時間點之後5s插入一個定時器，這樣就能形成每隔5秒調用一次callback2回調函數，可以用于周期性事件
    timer.SetTimer("callback2",, callback2, args)
    log.Println("callback2")
}

func main() {
    // cpu多核
    runtime.GOMAXPROCS(runtime.NumCPU())
    // 定時器1，傳入兩個參數
    timer.SetTimer("callback1",, callback1, []string{"hello", "world"})
    // 定時器2，不傳參數
    timer.SetTimer("callback2",, callback2, nil)
    // 移除定時器
    //timer.Delete(timer.TimerMap["callback2"])
    //運作計時器
    timer.Run()
}
           

完整的代碼在，https://github.com/liberalman/timer_server，這裡隻列出主要的部分。

将代碼下載下傳到您本地GOPATH路徑下，進入timer_server目錄下，執行

go build

用以編譯生成timer_server，沒問題的話，運作

./timer_server

看到結果

2015/04/13 13:06:43 第1s

2015/04/13 13:06:44 第2s

2015/04/13 13:06:45 第3s

2015/04/13 13:06:45 callback1(hello,world)

2015/04/13 13:06:46 第4s

2015/04/13 13:06:47 第5s

2015/04/13 13:06:48 第6s

2015/04/13 13:06:48 callback2

2015/04/13 13:06:49 第7s

2015/04/13 13:06:50 第8s

2015/04/13 13:06:51 第9s

2015/04/13 13:06:52 第10s

2015/04/13 13:06:53 第11s

2015/04/13 13:06:53 callback2

……

結果顯示第3秒的時候，調用了callback1回調函數，并且隻調用了一次，以後不再調用了。

在第6s的時候，調用callback2，并且以後每隔5秒都會調用該函數，成了周期。

然後我們将

這行的注釋去掉，看看運作結果

2015/04/13 13:09:31 第1s

2015/04/13 13:09:32 第2s

2015/04/13 13:09:33 第3s

2015/04/13 13:09:34 callback1(hello,world)

2015/04/13 13:09:35 第4s

2015/04/13 13:09:36 第5s

2015/04/13 13:09:37 第6s

2015/04/13 13:09:38 第7s

2015/04/13 13:09:39 第8s

2015/04/13 13:09:40 第9s

2015/04/13 13:09:41 第10s

2015/04/13 13:09:42 第11s

……

可以看到，callback2不再被調用了，因為把這個定時器跟删除了，根據它的名稱callback2找到其位置删除的。

始于2015-04-10，北京；更新至2016-06-13，杭州。