用Golang实现基于时间轮算法的定时器

关于什么是时间轮算法，自己百度吧，这里列出我用go语言实现的时间轮算法，已经上线应用，稳定。

package timer

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

const wheel_cnt uint8 = 5                                                                   //时间轮数量5个
var element_cnt_per_wheel = [wheel_cnt]uint32{256, 64, 64, 64, 64}                          //每个时间轮的槽(元素)数量。在 256+64+64+64+64 = 512 个槽中，表示的范围为 2^32
var right_shift_per_wheel = [wheel_cnt]uint32{8, 6, 6, 6, 6}                                //当指针指向当前时间轮最后一位数，再走一位就需要向上进位。每个时间轮进位的时候，使用右移的方式，最快实现进位。这里是每个轮的进位二进制位数
var base_per_wheel = [wheel_cnt]uint32{1, 256, 256 * 64, 256 * 64 * 64, 256 * 64 * 64 * 64} //记录每个时间轮指针当前指向的位置
var mutex sync.Mutex                                                                        //加锁
var rwmutex sync.RWMutex
var newest [wheel_cnt]uint32                           //每个时间轮当前指针所指向的位置
var timewheels [5][]*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 <= 0 {
        return
    }
    var bucket_no uint8 = 0
    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 = 1
        if bucket_no == 0 {
            tmp = 0
        }
        left -= base_per_wheel[bucket_no] * (element_cnt_per_wheel[bucket_no] - newest[bucket_no] - tmp)
        bucket_no++
    }
    if offset < 1 {
        return
    }
    if inteval < base_per_wheel[bucket_no]*offset {
        return
    }
    left -= base_per_wheel[bucket_no] * (offset - 1)
    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 = 0
        for bucket_no = 0; bucket_no < wheel_cnt; bucket_no++ {
            newest[bucket_no] = (newest[bucket_no] + 1) % 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 0 == bucket_no || 0 == inteval {
                            //dolist.PushBack(value) //执行自定义处理函数
                            go doSomething(args)
                        } else {
                            SetTimer(value.Name, inteval, doSomething, args) //重新插入计时器
                        }
                    }
                    Delete(firstElement) //删除定时器
                }
                firstElement = head.Next() //重新定位到链表第一个元素头
            }
            if 0 != newest[bucket_no] { //指针不是0，还未转回到原点，跳出。如果回到原点，则说明转完了一圈，需要向高位进位1，则继续循环入高位步进一步。
                break
            }
        }
        rwmutex.RUnlock()
    }
}

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

func init() { //初始化
    var bucket_no uint8 = 0
    for bucket_no = 0; bucket_no < wheel_cnt; bucket_no++ {
        var i uint32 = 0
        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[0]
        var str2 string = values[1]
        log.Println("callback1(" + str1 + "," + str2 + ")")
    } else {
        log.Println("callback1()")
    }
}

func callback2(args interface{}) {
    //每次在当前时间点之后5s插入一个定时器，这样就能形成每隔5秒调用一次callback2回调函数，可以用于周期性事件
    timer.SetTimer("callback2", 5, callback2, args)
    log.Println("callback2")
}

func main() {
    // cpu多核
    runtime.GOMAXPROCS(runtime.NumCPU())
    // 定时器1，传入两个参数
    timer.SetTimer("callback1", 3, callback1, []string{"hello", "world"})
    // 定时器2，不传参数
    timer.SetTimer("callback2", 6, 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秒都会调用该函数，成了周期。
然后我们将

//timer.Delete(timer.TimerMap["callback2"])

这行的注释去掉，看看运行结果
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，杭州。

用Golang实现基于时间轮算法的定时器

浏览过的版块