关于什么是时间轮算法,自己百度吧,这里列出我用go语言实现的时间轮算法,已经上线应用,稳定。
package timer
import (
"log"
"sync"
"time"
)
const wheel_cnt uint8 = 5 //时间轮数量5个
var element_cnt_per_wheel = [wheel_cnt]uint32{256, 64, 64} //每个时间轮的槽(元素)数量。在 256+64+64+64+64 = 512 个槽中,表示的范围为 2^32
var right_shift_per_wheel = [wheel_cnt]uint32{8, 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",nil)
// 移除定时器
//timer.Delete(timer.TimerMap["callback2"])
//运行计时器
timer.Run()
}
完整的代码在,https://github.com/socho1/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,杭州。