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
}

以上比较直观展示 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:因为 < 单个时间精度,表示这个任务已经过期,需要马上执行针对改变的 delaynew >= 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 = 0
func (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 / interval
pos = step % numSlots - 1
circle = (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定时/延迟任务内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • 如何用go-zero 实现中台系统

    最近发现golang社区里出了一个新星的微服务框架,来自好未来,光看这个名字,就很有奔头,之前,也只是玩过go-micro,其实真正的还没有在项目中运用过,只是觉得 微服务,grpc 这些很高大尚,还没有在项目中,真正的玩过,我看了一下官方提供的工具真的很好用,只需要定义好,舒适文件jia结构 都生成了,只需要关心业务, 加上最近 有个投票的活动,加上最近这几年中台也比较火,所以决定玩一下, 先聊聊中台架构思路吧,look 先看架 中台的概念大概就是把一个一个的app 统一起来,反正我是这样理解

  • 利用go-zero在Go中快速实现JWT认证的步骤详解

    关于JWT是什么,大家可以看看官网,一句话介绍下:是可以实现服务器无状态的鉴权认证方案,也是目前最流行的跨域认证解决方案. 要实现JWT认证,我们需要分成如下两个步骤 客户端获取JWT token. 服务器对客户端带来的JWT token认证. 1. 客户端获取JWT Token 我们定义一个协议供客户端调用获取JWT token,我们新建一个目录jwt然后在目录中执行 goctl api -o jwt.api,将生成的jwt.api改成如下: type JwtTokenRequest stru

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

    一个系统中存在着大量的调度任务,同时调度任务存在时间的滞后性,而大量的调度任务如果每一个都使用自己的调度器来管理任务的生命周期的话,浪费cpu的资源而且很低效. 本文来介绍 go-zero 中 延迟操作,它可能让开发者调度多个任务时,只需关注具体的业务执行函数和执行时间「立即或者延迟」.而 延迟操作,通常可以采用两个方案: Timer:定时器维护一个优先队列,到时间点执行,然后把需要执行的 task 存储在 map 中collection 中的 timingWheel ,维护一个存放任务组的数组

  • go-zero 如何应对海量定时/延迟任务

    一个系统中存在着大量的调度任务,同时调度任务存在时间的滞后性,而大量的调度任务如果每一个都使用自己的调度器来管理任务的生命周期的话,浪费cpu的资源而且很低效. 本文来介绍 go-zero 中 延迟操作,它可能让开发者调度多个任务时,只需关注具体的业务执行函数和执行时间「立即或者延迟」.而 延迟操作,通常可以采用两个方案: Timer:定时器维护一个优先队列,到时间点执行,然后把需要执行的 task 存储在 map 中collection 中的 timingWheel ,维护一个存放任务组的数组

  • 建立在Tablestore的Wifi设备监管系统架构实现

    Wifi设备监管 公司通过监管系统维护Wifi设备属性.采集Wifi设备监控数据.当需要Wifi设备上.下线时,通过监管系统操作完成设备的添加.下线,同时可通过系统修改.增加设备属性信息,如:设备mac地址.设备型号.设备地理位置等.设备上线后,会定期向系统推送监控数据,从而完成设备监控数据的采集.采集数据包含:cpu.内存.连接数.Wan口流量与流速.2.4G与5G模块的信道数据等. 通过分析监控数据指标.分析设备运行状态,动态将问题设备的运行状态修改为:预警.报警.借助系统,网络部门可以快速

  • 海量文件复制和备份的技巧(xcopy命令实现)

    网站服务器上的文件通常会越来越多,大量文件进行复制和备份的时候,使用Windows自带的复制粘贴功能,速度会非常慢,主要原因和Windows复制时候计算剩余时间有关,这给网站管理员带来了不少困扰.我这里就总结一个简单而快速的进行海量文件复制和备份的方法. 首先,Xcopy是什么? Xcopy是一个Windows自带的复制文件和目录的工具,可以复制子目录,是个非常好用的命令行工具. 常用的带子目录的复制方法是: xcopy "d:\website" "d:\webbak&quo

  • PowerShell小技巧之定时抓取屏幕图像

    昨天的博文写了定时记录操作系统行为,其实说白了就是抓取了击键的记录和对应窗口的标题栏,而很多应用程序标题栏又包含当时记录的文件路径和文件名,用这种方式可以大致记录操作了哪些程序,打开了哪些文件,以及敲击了哪些按键.事实上这样记录操作系统的行为显得相对单薄一点,因为记录的内容不太形象,对于新手来说太过于隐晦了,对于人类来说,图像会比文字更加有利于用户理解.当操作系统不方便装屏幕记录软件,但又需要看已经登录用户在干什么的时候,用PowerShell的脚本来实现定时抓取图像的方式记录操作,查看图像就知

  • Rsync实现文件定时同步技巧

    使用背景: 2005年5月,商务网处于上线阶段.但,由于总总原因一直迟迟不能上线.系统基本功能都已经实现了,但是作为一个商务网整体来说,文件同步这块还没有完成.正好自己这段时间不是很忙.对这块也有些兴趣,所以决定摸摸看. 万事开头难,从没接触过什么同步之类的,怎么开始呢?幸好以前做邮件系统的时候,认识了Eyou的小熊,请教了他,于是推荐了Rsync这个好工具给我.在此,特别感谢小熊的帮助.        Rsync的优点是部署方便,使用方便,效率高,并且它是开源的.目前在某些Linux服务器版中

  • 关于代码阅读问题的小技巧 脚本之家原创(适合所有网站)不定时更新

    因时间问题不可能一次写的很多,但我们会经常更新,方便大家.希望学习编程的朋友不要因为从网站复制了代码,而不能正常的运行而放弃.其实老鸟也是经常地遇到这类错误,看到错误提示就知道是什么问题,前期大家可以用google或百度搜下错误信息,就能知道大约的问题,慢慢培养对错误信息的认识,下次碰到一下就解决了. 1.字符的替换问题   很多网站都是将一些单引号(')替换为(')中文的单引号,直接导致代码无法运行   常见的字符有(",',$,&,*等) 2.复制代码的时候代码都是在一行显示的, 根

  • PowerShell小技巧之定时记录操作系统行为

    作为系统管理员,有些时候是需要记录系统中的其他用户的一些操作行为的,例如:当系统管理员怀疑系统存在漏洞,且已经有被植入后门或者创建隐藏账户时,就需要对曾经登陆的用户进行监控,保存其打开或者操作过的文件.或者在另外一个场景,当黑客拿下一个普通权限的shell之后,想看看最近有哪些用户登陆过,操作过什么,以便根据用户习惯采取进一步行动获取更高权限,这个时候记录用户行为就显得很重要了. 可能有读者觉得此时安装个监控软件不就行了么,拜托,你入侵别人的系统,你装个监控软件,你把管理员试做无物么?这个时候P

  • JavaScript小技巧整理篇(非常全)

    能够为大家提供这些简短而实用的JavaScript技巧来提高大家编程能力,这对于我来说是件很开心的事.每天仅花上不到2分钟的时间中,你将可以读遍JavaScript这门可怕的语言所呈现给我们的特性:performance(性能), conventions(协议), hacks(代码hack), interview questions(面试问题)及所有其他的项. #24 - 使用 === 代替 == ==(或者!=)做对比的时候会将进行对比的两者转换到同一类型再比较.===(或者!==)则不会,他

  • 批处理的高级运用技巧

    批处理的高级运用1 一.简单批处理内部命令  批处理的高级运用!!小技巧哦!!!  批处理大家都很熟悉吧!  下面介绍批处理的几个知识点:  1:巧用FC命令做查木马工具:  首先建立批处理文件atm.bat:写入代码:@echo off  dir c:\windows\system32\*.exe >c:??.txt  dir c:\windows\system32\*.dll >c:??.txt  2:建立批处理文件WLTS.bat 写入代码:@echo off  dir c:\windo

随机推荐