延时任务管理——层级时间轮 #11
Assignees
Labels
Comments
Open
JemmyH
added
Digging In
简单时间轮
层级时间轮
|
实现用到的数据结构：
在具体的实现(参考 kafka 时间轮的实现)上，kafka 的实现和上述原理还有些不同。 1. 时间轮的表示方式
2. 时钟的驱动方式
|
bucket
type bucket struct {
// 以当前时间为基准，再经过 expiration 后到期。如果为 -1，表示已经过期
expiration int64
mu sync.Mutex
// 存储具体任务的双链表
timers *list.List
}再来看 // newBucket 创建一个 bucket
func newBucket() *bucket {
return &bucket{
timers: list.New(),
expiration: -1,
}
}
// GetExpiration 返回当前 bucket 的 expiration
func (b *bucket) GetExpiration() int64 {
return atomic.LoadInt64(&b.expiration)
}
// SetExpiration 设置当前 bucket 的 expiration，如果设置的值和已有的不相同，返回 true
func (b *bucket) SetExpiration(expiration int64) bool {
return atomic.SwapInt64(&b.expiration, expiration) != expiration
}
// Add 添加任务到 bucket 中。Timer 表示一个具体的任务
func (b *bucket) Add(t *Timer) {
b.mu.Lock()
// 主要做的是将这个 Timer 放到 bucket 的双链表中，
// 并且绑定 Timer 和 bucket 之间的关系。
e := b.timers.PushBack(t)
t.setBucket(b)
t.element = e
b.mu.Unlock()
}
func (b *bucket) remove(t *Timer) bool {
if t.getBucket() != b {
// If remove is called from t.Stop, and this happens just after the timing wheel's goroutine has:
// 1. removed t from b (through b.Flush -> b.remove)
// 2. moved t from b to another bucket ab (through b.Flush -> b.remove and ab.Add)
// then t.getBucket will return nil for case 1, or ab (non-nil) for case 2.
// In either case, the returned value does not equal to b.
return false
}
b.timers.Remove(t.element)
t.setBucket(nil)
t.element = nil
return true
}
// Remove 将一个 Timer 从 bucket 的双链表中移除
func (b *bucket) Remove(t *Timer) bool {
b.mu.Lock()
defer b.mu.Unlock()
return b.remove(t)
}
// Flush 核心方法，bucket 对应的任务到期后，将所有的任务取出，并逐个执行传入的 reinsert 方法
func (b *bucket) Flush(reinsert func(*Timer)) {
var ts []*Timer
b.mu.Lock()
// 遍历双链表，取出后删除
for e := b.timers.Front(); e != nil; {
next := e.Next()
t := e.Value.(*Timer)
b.remove(t)
ts = append(ts, t)
e = next
}
b.mu.Unlock()
b.SetExpiration(-1) // TODO: Improve the coordination with b.Add()
// 逐个调用回调函数处理定时任务
for _, t := range ts {
reinsert(t)
}
}Timer
// Timer represents a single event. When the Timer expires, the given
// task will be executed.
type Timer struct {
expiration int64 // in milliseconds
// 到期执行的方法
task func()
// The bucket that holds the list to which this timer's element belongs.
//
// NOTE: This field may be updated and read concurrently,
// through Timer.Stop() and Bucket.Flush().
b unsafe.Pointer // type: *bucket
// The timer's element.
element *list.Element
}
func (t *Timer) getBucket() *bucket {
return (*bucket)(atomic.LoadPointer(&t.b))
}
func (t *Timer) setBucket(b *bucket) {
atomic.StorePointer(&t.b, unsafe.Pointer(b))
}
// Stop 将还未执行的 Timer 移除，如果移除成功返回 true，Timer 已经过期或者已经被停止，返回 false
// 考虑到层级时间轮的实现方式，一个 Timer 有可能会被转移到不同的 bucket 上，因此，为保险起见，遍历对应的 bucket
func (t *Timer) Stop() bool {
stopped := false
for b := t.getBucket(); b != nil; b = t.getBucket() {
// If b.Remove is called just after the timing wheel's goroutine has:
// 1. removed t from b (through b.Flush -> b.remove)
// 2. moved t from b to another bucket ab (through b.Flush -> b.remove and ab.Add)
// this may fail to remove t due to the change of t's bucket.
stopped = b.Remove(t)
// Thus, here we re-get t's possibly new bucket (nil for case 1, or ab (non-nil) for case 2),
// and retry until the bucket becomes nil, which indicates that t has finally been removed.
}
return stopped
}TimingWheel
type TimingWheel struct {
// 轮中每一个格的跨度，单位 毫秒
tick int64
// 格子个数
wheelSize int64
// 一轮总跨度，单位 毫秒
interval int64
// 当前指针时间，单位 毫秒。必须是 tick 的倍数
currentTime int64
// 格子
buckets []*bucket
// 延时队列
queen *DelayQueue
// 上级的时间轮
overflowWheel unsafe.Pointer // type: *TimingWheel
// 用于退出当前 TimingWheel
exitC chan struct{}
waitGroup waitGroupWrapper
}在实现具体的逻辑之前，我们定义以下的工具函数： // timeToMs 将 t 变成时间戳，单位为毫秒
func timeToMs(t time.Time) int64 {
return t.UnixNano() / int64(time.Millisecond)
}
// truncate 返回小于等于 x 的 m 的整数倍，如传入 (100,11)，最靠近且小于 100 的 11 的倍数是 99，所以返回 99
func truncate(x, m int64) int64 {
if m <= 0 {
return x
}
return x - x%m
}
// waitGroupWrapper 其实是对 sync.WaitGroup 的封装
type waitGroupWrapper struct {
sync.WaitGroup
}
func (w *waitGroupWrapper) Wrap(cb func()) {
w.Add(1)
go func() {
cb()
w.Done()
}()
}创建一个层级时间轮实例func NewTimingWheel(tick time.Duration, wheelSize int64) *TimingWheel {
// 将 tick 转化为毫秒
tickMs := int64(tick / time.Millisecond)
if tickMs <= 0 {
panic("tick must be greater than 0ms")
}
if wheelSize < 0 {
panic("wheelSize must be greater than 0")
}
startMs := timeToMs(time.Now())
return newTimingWheelInternal(tickMs, wheelSize, startMs, NewDelayQueen(int(wheelSize)))
}
func newTimingWheelInternal(tickMs int64, wheelSize int64, startMs int64, dq *DelayQueue) *TimingWheel {
// 创建并初始化 wheelSize 长度的 bucket
buckets := make([]*bucket, wheelSize)
for i := range buckets {
buckets[i] = newBucket()
}
return &TimingWheel{
tick: tickMs,
wheelSize: wheelSize,
interval: tickMs * wheelSize,
currentTime: truncate(startMs, tickMs),
buckets: buckets,
queen: dq,
exitC: make(chan struct{}),
waitGroup: waitGroupWrapper{sync.WaitGroup{}},
}
}时间轮启动func (tw *TimingWheel) Start() {
// 第一个 worker，不断从优先级队列中取出到期的元素，将其放在 queen.C 中
tw.waitGroup.Wrap(func() {
tw.queen.Poll(tw.exitC, func() int64 {
return timeToMs(time.Now().UTC())
})
})
tw.waitGroup.Wrap(func() {
for {
select {
case item := <-tw.queen.C:
// b 表示到达执行时间的时间格
b := item.(*bucket)
// 时间轮会将 currentTime 往前移动到 bucket到期的时间 或 下一个时间格
tw.advanceClock(b.GetExpiration())
// 取出bucket队列的数据，并调用 addOrRun 方法执行
b.Flush(tw.addOrRun)
case <-tw.exitC:
// 退出时间轮
return
}
}
})
}
// advanceClock 表示有 bucket 事件到期，需要更新 tw.currentTime 为 bucket 的 expiration
func (tw *TimingWheel) advanceClock(expiration int64) {
curTime := atomic.LoadInt64(&tw.currentTime)
// 过期时间大于等于（当前时间+tick）
if expiration >= curTime+tw.tick {
// 更新 tw.currentTime 为 expiration，从而推动时间轮的转动
curTime = truncate(expiration, tw.tick)
atomic.StoreInt64(&tw.currentTime, curTime)
// 如果有上层时间轮，也要递归处理上层时间轮的推动
overflowWheel := atomic.LoadPointer(&tw.overflowWheel)
if overflowWheel != nil {
(*TimingWheel)(overflowWheel).advanceClock(curTime)
}
}
}之后调用 func (tw *TimingWheel) addOrRun(t *Timer) {
if !tw.add(t) {
go t.task()
}
}首先调用 添加一个任务到时间轮核心实现在
// add 尝试将 t 添加到时间轮中，如果成功添加，说明 t 还没到执行的时间，插入到了某个 bucket，返回 true；如果返回 false，说明 t 已经到
// 执行的时间了，没必要添加到时间轮中
func (tw *TimingWheel) add(t *Timer) bool {
currentTime := atomic.LoadInt64(&tw.currentTime)
if t.expiration < currentTime+tw.tick {
// case A: 已经过期或在当前时间格执行
return false
} else if t.expiration < currentTime+tw.interval {
// case B:到期时间在第一层时间轮之内，这里也包含上层时间轮的降级
virtualID := t.expiration / tw.tick
b := tw.buckets[virtualID%tw.wheelSize]
b.Add(t)
if b.SetExpiration(virtualID * tw.tick) {
tw.queen.Offer(b, b.expiration)
}
return true
} else {
// case C: 在更上层的时间轮中(过期时间超过了 tw.interval)
overflowWheel := atomic.LoadPointer(&tw.overflowWheel)
if overflowWheel == nil {
// 创建一个上层时间轮，注意这里创建的 TimingWheel 的 tick 为 底下一层的 interval
atomic.CompareAndSwapPointer(
&tw.overflowWheel,
nil,
unsafe.Pointer(newTimingWheelInternal(tw.interval, tw.wheelSize, currentTime, tw.queen)),
)
overflowWheel = atomic.LoadPointer(&tw.overflowWheel)
}
// 递归调用，将这个 Timer 插入到更上层时间轮的某个 bucket 中
return (*TimingWheel)(overflowWheel).add(t)
}
}手动添加延时任务有了以上的逻辑，添加一个延时任务就变得非常简单：创建一个 Timer，设定好 func (tw *TimingWheel) AfterFunc(d time.Duration, h func()) *Timer {
t := &Timer{
expiration: timeToMs(time.Now().UTC().Add(d)),
task: h,
}
tw.addOrRun(t)
return t
}停止时间轮给信号给 func (tw *TimingWheel) Stop() {
close(tw.exitC)
tw.waitGroup.Wait()
} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
为什么会有 时间轮 这个概念？先看一个场景：在商品抢购项目中，短时间内会创建几百万个定时任务，创建的时候更新某个参数，一段时间以后再去核对，将与逻辑不符合的任务处理。
Golang 内置的 Timer 是采用最小堆来实现的，创建和删除的时间复杂度都为
O(log n)。如果有上万的连接，每个连接都会有很多超时任务，比如发送超时、心跳检测等，如果每个超时任务都对应一个Timer，性能会比较低下。论文 Hashed and Hierarchical Timing Wheels 提出了一种用于实现
Timer的高效数据结构：时间轮。采用时间轮实现的Timer，创建和删除的时间复杂度为O(1)。本次借助已有的实现 RussellLuo/timingwheel: Golang implementation of Hierarchical Timing Wheels. 说明原理。该作者的实现中并没有按照常规的使用类似钟表那样的环形数组，而是参考了
kafka的实现，在这个实现中，需要用到两个组件：本 issue 的目标是明白层级时间轮的运作原理以及从源码层面上理解上述项目实现的细节。
The text was updated successfully, but these errors were encountered: