go-zero 应对海量定时/延迟任务的技巧

一个系统中存在着大量的调度任务,同时调度任务存在时间的滞后性,而大量的调度任务如果每一个都使用自己的调度器来管理任务的生命周期的话,浪费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

首先我们先来在 collectioncache 中关于 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的过期处理,参数依次代表:

  • interval:时间划分刻度
  • numSlots:时间槽
  • execute:时间点执行函数

cache 中执行函数则是 删除过期key,而这个过期则由 timingWheel 来控制推进时间。

接下来,就通过 cachetimingWheel 的使用来认识。

初始化

// 真正做初始化
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
}

go-zero 应对海量定时/延迟任务的技巧

以上比较直观展示 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 执行。

go-zero 应对海量定时/延迟任务的技巧

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

go-zero 应对海量定时/延迟任务的技巧

扫一扫手机访问