时间:2021-05-22
一个系统中存在着大量的调度任务,同时调度任务存在时间的滞后性,而大量的调度任务如果每一个都使用自己的调度器来管理任务的生命周期的话,浪费cpu的资源而且很低效。
本文来介绍 go-zero 中 延迟操作,它可能让开发者调度多个任务时,只需关注具体的业务执行函数和执行时间「立即或者延迟」。而 延迟操作,通常可以采用两个方案:
Timer:定时器维护一个优先队列,到时间点执行,然后把需要执行的 task 存储在 map 中collection 中的 timingWheel ,维护一个存放任务组的数组,每一个槽都维护一个存储task的双向链表。开始执行时,计时器每隔指定时间执行一个槽里面的tasks。
方案2把维护task从 优先队列 O(nlog(n)) 降到 双向链表 O(1),而执行task也只要轮询一个时间点的tasks O(N),不需要像优先队列,放入和删除元素 O(nlog(n))。
我们先看看 go-zero 中自己对 timingWheel 的使用 :
cache 中的 timingWheel
首先我们先来在 collection 的 cache 中关于 timingWheel 的使用:
timingWheel, err := NewTimingWheel(time.Second, slots, func(k, v interface{}) { key, ok := k.(string) if !ok { return } cache.Del(key)})if err != nil { return nil, err}cache.timingWheel = timingWheel这是 cache 初始化中也同时初始化 timingWheel 做key的过期处理,参数依次代表:
在 cache 中执行函数则是 删除过期key,而这个过期则由 timingWheel 来控制推进时间。
接下来,就通过 cache 对 timingWheel 的使用来认识。
初始化
// 真正做初始化func newTimingWheelWithClock(interval time.Duration, numSlots int, execute Execute, ticker timex.Ticker) ( *TimingWheel, error) { tw := &TimingWheel{ interval: interval, // 单个时间格时间间隔 ticker: ticker, // 定时器,做时间推动,以interval为单位推进 slots: make([]*list.List, numSlots), // 时间轮 timers: NewSafeMap(), // 存储task{key, value}的map [执行execute所需要的参数] tickedPos: numSlots - 1, // at previous virtual circle execute: execute, // 执行函数 numSlots: numSlots, // 初始化 slots num setChannel: make(chan timingEntry), // 以下几个channel是做task传递的 moveChannel: make(chan baseEntry), removeChannel: make(chan interface{}), drainChannel: make(chan func(key, value interface{})), stopChannel: make(chan lang.PlaceholderType), } // 把 slot 中存储的 list 全部准备好 tw.initSlots() // 开启异步协程,使用 channel 来做task通信和传递 go tw.run() return tw, nil}以上比较直观展示 timingWheel 的 “时间轮”,后面会围绕这张图解释其中推进的细节。
go tw.run() 开一个协程做时间推动:
func (tw *TimingWheel) run() { for { select { // 定时器做时间推动 -> scanAndRunTasks() case <-tw.ticker.Chan(): tw.onTick() // add task 会往 setChannel 输入task case task := <-tw.setChannel: tw.setTask(&task) ... } }}可以看出,在初始化的时候就开始了 timer 执行,并以internal时间段转动,然后底层不停的获取来自 slot 中的 list 的task,交给 execute 执行。
Task Operation
紧接着就是设置 cache key :
func (c *Cache) Set(key string, value interface{}) { c.lock.Lock() _, ok := c.data[key] c.data[key] = value c.lruCache.add(key) c.lock.Unlock() expiry := c.unstableExpiry.AroundDuration(c.expire) if ok { c.timingWheel.MoveTimer(key, expiry) } else { c.timingWheel.SetTimer(key, value, expiry) }}先看在 data map 中有没有存在这个key存在,则更新 expire -> MoveTimer()第一次设置key -> SetTimer()
所以对于 timingWheel 的使用上就清晰了,开发者根据需求可以 add 或是 update。
同时我们跟源码进去会发现:SetTimer() MoveTimer() 都是将task输送到channel,由 run() 中开启的协程不断取出 channel 的task操作。
SetTimer() -> setTask():
not exist task:getPostion -> pushBack to list -> setPositionexist task:get from timers -> moveTask()
MoveTimer() -> moveTask()
由上面的调用链,有一个都会调用的函数:moveTask()
func (tw *TimingWheel) moveTask(task baseEntry) { // timers: Map => 通过key获取 [positionEntry「pos, task」] val, ok := tw.timers.Get(task.key) if !ok { return } timer := val.(*positionEntry) // {delay < interval} => 延迟时间比一个时间格间隔还小,没有更小的刻度,说明任务应该立即执行 if task.delay < tw.interval { threading.GoSafe(func() { tw.execute(timer.item.key, timer.item.value) }) return } // 如果 > interval,则通过 延迟时间delay 计算其出时间轮中的 new pos, circle pos, circle := tw.getPositionAndCircle(task.delay) if pos >= timer.pos { timer.item.circle = circle // 记录前后的移动offset。为了后面过程重新入队 timer.item.diff = pos - timer.pos } else if circle > 0 { // 转移到下一层,将 circle 转换为 diff 一部分 circle-- timer.item.circle = circle // 因为是一个数组,要加上 numSlots [也就是相当于要走到下一层] timer.item.diff = tw.numSlots + pos - timer.pos } else { // 如果 offset 提前了,此时 task 也还在第一层 // 标记删除老的 task,并重新入队,等待被执行 timer.item.removed = true newItem := &timingEntry{ baseEntry: task, value: timer.item.value, } tw.slots[pos].PushBack(newItem) tw.setTimerPosition(pos, newItem) }}以上过程有以下几种情况:
delay < internal:因为 < 单个时间精度,表示这个任务已经过期,需要马上执行针对改变的 delay:new >= old:<newPos, newCircle, diff>newCircle > 0:计算diff,并将 circle 转换为 下一层,故diff + numslots如果只是单纯延迟时间缩短,则将老的task标记删除,重新加入list,等待下一轮loop被execute
Execute
之前在初始化中,run() 中定时器的不断推进,推进的过程主要就是把 list中的 task 传给执行的 execute func。我们从定时器的执行开始看:
// 定时器 「每隔 internal 会执行一次」func (tw *TimingWheel) onTick() { // 每次执行更新一下当前执行 tick 位置 tw.tickedPos = (tw.tickedPos + 1) % tw.numSlots // 获取此时 tick位置 中的存储task的双向链表 l := tw.slots[tw.tickedPos] tw.scanAndRunTasks(l)}紧接着是如何去执行 execute:
func (tw *TimingWheel) scanAndRunTasks(l *list.List) { // 存储目前需要执行的task{key, value} [execute所需要的参数,依次传递给execute执行] var tasks []timingTask for e := l.Front(); e != nil; { task := e.Value.(*timingEntry) // 标记删除,在 scan 中做真正的删除 「删除map的data」 if task.removed { next := e.Next() l.Remove(e) tw.timers.Del(task.key) e = next continue } else if task.circle > 0 { // 当前执行点已经过期,但是同时不在第一层,所以当前层即然已经完成了,就会降到下一层 // 但是并没有修改 pos task.circle-- e = e.Next() continue } else if task.diff > 0 { // 因为之前已经标注了diff,需要再进入队列 next := e.Next() l.Remove(e) pos := (tw.tickedPos + task.diff) % tw.numSlots tw.slots[pos].PushBack(task) tw.setTimerPosition(pos, task) task.diff = 0 e = next continue } // 以上的情况都是不能执行的情况,能够执行的会被加入tasks中 tasks = append(tasks, timingTask{ key: task.key, value: task.value, }) next := e.Next() l.Remove(e) tw.timers.Del(task.key) e = next } // for range tasks,然后把每个 task->execute 执行即可 tw.runTasks(tasks)}具体的分支情况在注释中说明了,在看的时候可以和前面的 moveTask() 结合起来,其中 circle 下降,diff 的计算是关联两个函数的重点。
至于 diff 计算就涉及到 pos, circle 的计算:
// interval: 4min, d: 60min, numSlots: 16, tickedPos = 15// step = 15, pos = 14, circle = 0func (tw *TimingWheel) getPositionAndCircle(d time.Duration) (pos int, circle int) { steps := int(d / tw.interval) pos = (tw.tickedPos + steps) % tw.numSlots circle = (steps - 1) / tw.numSlots return}上面的过程可以简化成下面:
steps = d / intervalpos = step % numSlots - 1circle = (step - 1) / numSlots总结
timingWheel 靠定时器推动,时间前进的同时会取出当前时间格中 list「双向链表」的task,传递到 execute 中执行。因为是是靠 internal 固定时间刻度推进,可能就会出现:一个 60s 的task,internal = 1s,这样就会空跑59次loop。
而在扩展时间上,采取 circle 分层,这样就可以不断复用原有的 numSlots ,因为定时器在不断 loop,而执行可以把上层的 slot 下降到下层,在不断 loop 中就可以执行到上层的task。这样的设计可以在不创造额外的数据结构,突破长时间的限制。
同时在 go-zero 中还有很多实用的组件工具,用好工具对于提升服务性能和开发效率都有很大的帮助,希望本篇文章能给大家带来一些收获。
项目地址
https://github.com/tal-tech/go-zero
好未来技术
到此这篇关于go-zero 如何应对海量定时/延迟任务的文章就介绍到这了,更多相关go-zero定时/延迟任务内容请搜索以前的文章或继续浏览下面的相关文章希望大家以后多多支持!
声明:本页内容来源网络,仅供用户参考;我单位不保证亦不表示资料全面及准确无误,也不保证亦不表示这些资料为最新信息,如因任何原因,本网内容或者用户因倚赖本网内容造成任何损失或损害,我单位将不会负任何法律责任。如涉及版权问题,请提交至online#300.cn邮箱联系删除。
定时器是什么Golang原生time包下可以用来执行一些定时任务或者是周期性的任务的一个工具本文基于Go1.14,如果以下文章有哪里不对或者问题的地方,欢迎讨论
前言定时器线程池提供了定时执行任务的能力,即可以延迟执行,可以周期性执行。但定时器线程池也还是线程池,最底层实现还是ThreadPoolExecutor,可以参
记录下Spring自带的定时任务用法。spring中使用定时任务基于xml配置文件使用定时任务首先配置spring开启定时任务
定时任务在项目中经常会使用到,本文主要根据博主自己使用定时的经验分如下几点介绍定时任务:1、Quartz定时任务简介及Spring配置Quartz定时任务2、S
@schedule注解是springboot常用的定时任务注解,使用起来简单方便,但是如果定时任务非常多,或者有的任务很耗时,会影响到其他定时任务的执行,因为s