用于提早工作的调度。
场景
设想这么一个场景，我须要保护一个连接池，连接池中的连贯是有超时工夫的，连贯会放弃肯定频率发送心跳包比方5s发一次，30s内如果没有收到keep-alive就会过期，到过期工夫的conn断开连接，如何去设计这个连接池？
场景形象
这个场景能够了解为，我收到一个申请之后，在提早30秒后须要执行一个动作，并且如果在30s内收到同样的申请，就把这个工作再推延30s，咱们应该怎么做呢？
可行解
那假如我的连接池最大连接数是1000，简略的办法是每个连贯保护一个最近的keep-alive工夫，启动一个定时器，每秒去遍历一次所有连贯，到工夫了就断开连接。收到心跳时，更新这个连贯的工夫，那如果有10000个连贯呢？每秒就要遍历10000次能力确定哪些工作要删掉，这种形式是很浪费资源的。
更优解
应用工夫轮算法，工夫轮算法的根本思维是将提早工作扩散，不在一个中央去保护，防止因为都放在一个中央，每次都要进行遍历的损失。
怎么把工作进行扩散呢？依照提早工作的最大工夫限度和timer执行的工夫，比方说我最大反对提早60秒，每秒都扫一次，那我就能够建设一个数组叫做timeWheel，timeWheel上放的是这个工夫到期的连贯组成的数组connList，保护一个cur_index，指向timeWheel的某一个index。
执行的流程是从这个index开始向后遍历，每秒挪动一个index，到最大index的时候返回到0持续循环，每次挪动都断开index外面所有连贯。
接管到心跳时，把这个连贯从以后的timeWheel的index中去掉，index减少30后对60取模保障不越界。
优化：尽管每秒做删除的时候不必扫数组了，然而这样的话，收到心跳的时候就要扫数组了，那怎么优化收到心跳时扫数组的操作呢？从timeWheel取连贯的过程能够通过map做conn到timeWheel中index的索引，不用查找整个工夫轮。在conn批改timeWheel的index，
图解
缺点
这个场景中，最大延迟时间是60s，所以一个大小为60的timeWheel就够了，那如果是一天呢？一天是86400s，就须要用到86400大小的timeWheel了。能够通过分层来做优化。
多层工夫轮多层工夫轮是单层工夫轮的优化计划，用于缓解单层工夫轮在工夫范畴大，跨度小的状况下，timeWheel的大小也须要增大的状况。
场景
把下面的场景的工夫范畴裁减到7天，86400s*7
怎么分层
对于这种场景，咱们能够分为4层工夫轮。
第一层timeWheel1示意秒，大小为60，每个元素示意1s。
第二层timeWheel2示意分，大小为60，每个元素示意1m。
第三层timeWheel3示意小时，大小为24，每个元素示意1h。
第四层timeWheel4示意天，大小为7，每个元素示意1d。
执行流程
前提设定
假如以后timeWheel的状态如下
timeWheel1的60个index都为空，cur_index1在index0地位
timeWheel2的60个index都为空，cur_index2在index0地位
timeWheel3的24个index都为空，cur_index3在index0地位
timeWheel4的7歌index都为空，cur_index4在index0地位
工作增加流程
接管到一个工作1，须要提早7100s之后执行，
计算是否能够落到timeWheel1中。7100比timeWheel1的最大范畴60大，不能放在timeWheel1中。
计算是否能够落到timeWheel2中。7100/60 = 118.333向下取整118，118比timeWheel2的最大范畴60大。不能放在timeWheel2中。
计算是否能够落到timeWheel3中。$7100/(60*60) = 1.97222$，1.97222向下取整得1，1比timeWheel3的最大范畴24小，能够放在timeWheel3中。计算下一级残余多少工夫：$7100 % (60*60) = 3500$，还剩下3500s，存储在cur_index3+1这个地位上，同时保留残余的工夫3500。
这也就是工夫轮的降级过程。
工夫轮运行流程
运行的时候波及到工夫轮的降级过程。
以四层环为例
一个简略暴力的实现思路是，提前定义有多少层环，有多少层环，就开多少个定时器。
下层定时器到工夫，把对应cur_index中的工作放到上层定时器中，这种须要进行这几个环的一个解决程序做同步，不然可能呈现下层放到上层的时候，上层曾经走过了这个地位的状况。
另一个实现形式是只有一个定时器，先走最底层，最底层走完一圈的时候走下层，通过下层这个cur_index中的工作的剩余时间来判断，没有剩余时间的立刻执行，有剩余时间的放到底层的。如果这一层也跑了一圈，再去上一层取工作。以此类推，实现较为简单。
go-zero中的实现是2层循环，能够保护了一个底层slot和一个循环圈数circle，每次扫描的数量尽管多了，然而实现简略，能够在上面的源码中参考一下。
工夫轮在go中的实现
Go语言中工夫轮的实现
齐全兼容golang定时器的高性能工夫轮实现(go-timewheel)
go-zero中工夫轮的实现
go-zero中的timeWheel用于程序外部缓存cache的过期清理操作。以下只关怀工夫轮的实现形式。
// 工夫轮构造体
type TimingWheel struct {
interval time.Duration //工夫轮每个槽位的工夫距离
ticker timex.Ticker //定时器
slots []*list.List //槽位数组
timers *SafeMap //一个避免内存透露的map，用来装什么的？
tickedPos int //当初指向的槽位
numSlots int //槽数
execute Execute //到工夫执行的函数
// 对外提供的办法，通过channel来与timeWheel交互
// 这种交互方式的益处大略是对于不关怀后果的调用方起到解耦成果，不须要同步期待
setChannel chan timingEntry //设置接口
moveChannel chan baseEntry //挪动接口
removeChannel chan interface{} //删除接口
drainChannel chan func(key, value interface{}) //
stopChannel chan lang.PlaceholderType //完结接口
}
//工夫轮元素构造体
type timingEntry struct {
baseEntry
value interface{}
circle int
diff int
removed bool
}
//根底字段
type baseEntry struct {
delay time.Duration
key interface{}
}
//key到slot的映射，以及一个timingEntry元素指针，
//用于通过key在timers中疾速查找到元素
type positionEntry struct {
pos int
item *timingEntry
}
//提早工作字段
type timingTask struct {
key interface{}
value interface{}
}
//办法定义
//以下办法是对外提供的办法，通过channel发送信号，run办法中监听各个channel，收到信号执行相应的办法
//立即执行所有工作
func (tw *TimingWheel) Drain(fn func(key, value interface{}))
//工作没到工夫就把工作的延迟时间更新。到工夫就间接执行。
func (tw *TimingWheel) MoveTimer(key interface{}, delay time.Duration)
//移除一个工作
func (tw *TimingWheel) RemoveTimer(key interface{})
//减少一个工作
func (tw *TimingWheel) SetTimer(key, value interface{}, delay time.Duration)
//进行工夫轮
func (tw *TimingWheel) Stop()
//以下办法是真正执行的办法
func (tw *TimingWheel) drainAll(fn func(key, value interface{}))
//pos用 延迟时间/槽位示意的工夫 先计算出往后数第几个槽，思考溢出和以后的槽位偏移，最终的pos = (tickedPos+d/interval)%numSlots
//circle = (d/interval-1)/numSlots
func (tw *TimingWheel) getPositionAndCircle(d time.Duration) (pos, circle int)
func (tw *TimingWheel) initSlots()
func (tw *TimingWheel) moveTask(task baseEntry)
func (tw *TimingWheel) onTick()
func (tw *TimingWheel) removeTask(key interface{})
func (tw *TimingWheel) removeTask(key interface{})
func (tw *TimingWheel) run()
func (tw *TimingWheel) runTasks(tasks []timingTask)
func (tw *TimingWheel) scanAndRunTasks(l *list.List)
//
func newTimingWheelWithClock(interval time.Duration, numSlots int, execute Execute, ticker timex.Ticker) (
*TimingWheel, error) {
tw := &TimingWheel{
interval: interval,
ticker: ticker,
slots: make([]*list.List, numSlots),
timers: NewSafeMap(),
tickedPos: numSlots - 1, // at previous virtual circle
execute: execute,
numSlots: numSlots,
setChannel: make(chan timingEntry),
moveChannel: make(chan baseEntry),
removeChannel: make(chan interface{}),
drainChannel: make(chan func(key, value interface{})),
stopChannel: make(chan lang.PlaceholderType),
}
tw.initSlots()
go tw.run()
return tw, nil
}
// Drain drains all items and executes them.
func (tw *TimingWheel) Drain(fn func(key, value interface{})) {
tw.drainChannel <- fn
}
//run办法，监听所有channel，在newTimeWheel的时候就启动了。
func (tw *TimingWheel) run() {
for {
select {
case <-tw.ticker.Chan(): //到工夫，执行对应slot上须要执行的工作。
tw.onTick()
case task := <-tw.setChannel: //往工夫轮上增加一个工作
tw.setTask(&task)
case key := <-tw.removeChannel: //从工夫轮上删除一个工作
tw.removeTask(key)
case task := <-tw.moveChannel: //更新一个工夫轮上工作的执行工夫
tw.moveTask(task)
case fn := <-tw.drainChannel:
tw.drainAll(fn)
case <-tw.stopChannel:
tw.ticker.Stop()
return
}
}
}
//先看减少，也就是setTask办法
//调用：
// run -> setTask
//逻辑：
// 从map索引中确定，这个工作是否曾经存在了
// 存在的话就通过moveTask挪动这个工作的地位
// 不存在的话，就计算出工作在环中绝对于以后的ticked的定位，以及要转的圈数circle，将工作放在环上，并且保护map索引
func (tw *TimingWheel) setTask(task *timingEntry) {
if task.delay < tw.interval {
task.delay = tw.interval
}
if val, ok := tw.timers.Get(task.key); ok {
entry := val.(*positionEntry)
entry.item.value = task.value
tw.moveTask(task.baseEntry)
} else {
pos, circle := tw.getPositionAndCircle(task.delay)
task.circle = circle
tw.slots[pos].PushBack(task)
tw.setTimerPosition(pos, task)
}
}
// 减少看完了，再看一下是怎么执行的，假如曾经扫到了这个工作所在的slot，
// 先保护一下曾经扫到的地位，而后从slot中拿出对应的list，扔到scanAndRunTask办法中执行。
func (tw *TimingWheel) onTick() {
tw.tickedPos = (tw.tickedPos + 1) % tw.numSlots
l := tw.slots[tw.tickedPos]
tw.scanAndRunTasks(l)
}
// 次要看scanAndRunTask办法，这个办法是真正在拿到list之后做的操作
//逻辑：
// 遍历整个list，先革除被删掉工作，再将循环圈数不为0的工作的圈数-1，
// 剩下的是圈数为0的无效工作，思考到有更新操作，更新操作的pos会推延到真正要执行的时候做，所以还要依据diff再看一下是不是一个被更新的操作。
// 最初后面这些都被过滤掉，剩下来的工作就是这次scan要执行的工作，把他们退出到执行队列中，通过runTask办法并发执行，这个办法中会管制并发数。
func (tw *TimingWheel) scanAndRunTasks(l *list.List) {
var tasks []timingTask
for e := l.Front(); e != nil; {
task := e.Value.(*timingEntry)
if task.removed {
next := e.Next()
l.Remove(e)
e = next
continue
} else if task.circle > 0 {
task.circle--
e = e.Next()
continue
} else if task.diff > 0 {
next := e.Next()
l.Remove(e)
// (tw.tickedPos+task.diff)%tw.numSlots
// cannot be the same value of tw.tickedPos
pos := (tw.tickedPos + task.diff) % tw.numSlots
tw.slots[pos].PushBack(task)
tw.setTimerPosition(pos, task)
task.diff = 0
e = next
continue
}
tasks = append(tasks, timingTask{
key: task.key,
value: task.value,
})
next := e.Next()
l.Remove(e)
tw.timers.Del(task.key)
e = next
}
tw.runTasks(tasks)
}
// 更新timeWheel中已存在的工作的延迟时间
// 调用：
// run -> setTask -> moveTask 在setTask中判断如果有这个key就moveTask
// run -> moveTask
func (tw *TimingWheel) moveTask(task baseEntry) {
val, ok := tw.timers.Get(task.key)
if !ok {
return
}
timer := val.(*positionEntry)
//如果task设置的延迟时间太小了，那就间接执行
if task.delay < tw.interval {
threading.GoSafe(func() {
tw.execute(timer.item.key, timer.item.value)
})
return
}
// 没到工夫，须要扭转地位，依据新的延迟时间计算出新的定位和circle
pos, circle := tw.getPositionAndCircle(task.delay)
//依据pos和circle还有旧数据，批改task的信息，做一些标记，在扫描到这个task的时候再真正批改和从新定位。
//提早这些更改的益处是，如果某些key频繁改变的话，不须要频繁进行重定位操作，重定位操作须要保障并发平安。
if pos >= timer.pos {
//新pos大于等于旧pos，更新工作的circle，pos还是用旧的，而是把工作的diff更新为新pos-旧pos，这里是为什么？
//思考场景，先触发tick，再触发move，因为tick运行工作的时候是go进来的，go进来的工作正在执行，这时候如果来move申请，就会有并发问题
//这里记录他的diff是标记曾经被改过了，下次跑到这个工作的时候就会触发pos更新。
timer.item.circle = circle
timer.item.diff = pos - timer.pos
} else if circle > 0 {
//pos提前了，不在这一圈触发，得算一下diff偏移量和走多少圈
circle-- //把diff的一圈扣掉
timer.item.circle = circle
//算diff把circle扣的一圈加回来 假如新pos是1，旧pos是2，num是5，diff就是4，示意 2+4=6，6%5=1，计算的时候，这里就会计算为1
timer.item.diff = tw.numSlots + pos - timer.pos
} else {
//pos提前了，并且就是这次循环，删除旧的增加新的
// 这里是不是没有思考到并发的状况？如果正在执行这个工作，那这里会不会有问题？
// 如果思考就算执行完，也要生成新的继续执行，是正当的。
// 不思考并发，不思考雷同key只执行一次，这里是能够的，如果要思考的话，这里可能会被执行两次
timer.item.removed = true
newItem := &timingEntry{
baseEntry: task,
value: timer.item.value,
}
tw.slots[pos].PushBack(newItem)
tw.setTimerPosition(pos, newItem)
}
}
//最初看一下是怎么删除的，其实这外面只是通过索引找到这个task，而后把task标记为删除。
//下面看scanAndRunTask办法的时候曾经看到了解决被删除的节点的逻辑
func (tw *TimingWheel) removeTask(key interface{}) {
val, ok := tw.timers.Get(key)
if !ok {
return
}
timer := val.(*positionEntry)
timer.item.removed = true
tw.timers.Del(key)
}