个人中心
文章发布
退出
作者热门文章
- ubuntu12.04环境下使用kvm ioctl接口实现最简单的虚拟机
- Ubuntu 通过无线网络安装Ubuntu Server启动系统后连接无线网络的方法
- 在Ubuntu上搭建网桥的方法
- ubuntu 虚拟机上网方式及相关配置详解
26
4
CFSDN坚持开源创造价值，我们致力于搭建一个资源共享平台，让每一个IT人在这里找到属于你的精彩世界.
这篇CFSDN的博客文章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 。
首先我们先来在 collection 的 cache 中关于 timingWheel 的使用:
|
1
2
3
4
5
6
7
8
9
10
11
12
|
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 来控制推进时间.
接下来，就通过 cache 对 timingWheel 的使用来认识.
初始化 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
// 真正做初始化
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() 开一个协程做时间推动:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
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 :
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
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() 。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
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。我们从定时器的执行开始看:
|
1
2
3
4
5
6
7
8
|
// 定时器 「每隔 internal 会执行一次」
func (tw *TimingWheel) onTick() {
// 每次执行更新一下当前执行 tick 位置
tw.tickedPos = (tw.tickedPos +
1
) % tw.numSlots
// 获取此时 tick位置 中的存储task的双向链表
l := tw.slots[tw.tickedPos]
tw.scanAndRunTasks(l)
}
|
紧接着是如何去执行 execute:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
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 的计算:
|
1
2
3
4
5
6
7
8
|
// 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
}
|
上面的过程可以简化成下面:
|
1
2
3
|
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定时/延迟任务内容请搜索我以前的文章或继续浏览下面的相关文章希望大家以后多多支持我！ 。
原文链接：https://www.cnblogs.com/kevinwan/p/13825605.html 。
最后此篇关于go-zero 应对海量定时/延迟任务的技巧的文章就讲到这里了,如果你想了解更多关于go-zero 应对海量定时/延迟任务的技巧的内容请搜索CFSDN的文章或继续浏览相关文章，希望大家以后支持我的博客！ 。
26
4
0
我正在使用一个简单的脚本来延迟加载页面上的所有图像；图像源的路径包含在 data-src 属性中，然后放入 img 标记的实际 src 属性中。几乎大多数(？)延迟加载方法的实现都是如何工作的。 这是
我有一个具有多层 (SKNodes) 背景、游戏层、前景和 HUD 的场景，每个场景中都有多个 SKSpriteNode，用于滚动和您可以收集和点击的对象。 hud 层只有一个 SKSpriteNod
我有一个 Controller 函数来创建一些东西。调用该函数时，将运行 setInterval 来获取项目的状态。 这是服务: (function () { 'use strict';
在我的应用程序中，我播放音频直播，延迟非常重要。我正在使用 AVPlayer，但启动需要 5-6 秒，并且我需要最多 3 秒的延迟。我怎样才能更快地开始播放并减少延迟？设置一个小缓冲区就可以了？如何使
我有一个恼人的问题。我有这个简单的服务器代码(比方说): #!/usr/bin/env python3 import wsgiref.simple_server def my_func(env, st
我是 jquery deferreds 的新手。这里我有一个简单的example 。 谁能告诉我为什么在其他函数完成之前就触发完成函数(“现在是我的时间”)？ 这里的人 example还创建一个延迟对
正在放置关闭 之前的标签标记相同的 sa 将它们放在 中部分并指定 defer="defer"属性？ 最佳答案 是/否。 是的，因为放置 defer 标签会等到文档加载完毕后再执行。 否，因为放置
我知道Javascript没有delay(500)方法，它会延迟执行500毫秒，所以我一直试图通过使用setTimeout和setInterval来解决这个问题。 for(var i =0; i< 1
我们有一个读写主服务器和复制的从读服务器。在某些网络用例中，数据被发布并立即读取以发送回服务器。立即读取是在读取从属设备上完成的，由于延迟，数据尚未在那里更新。 我知道这可能是复制设置的一个常见问题，
我有以下 dag 设置以从 2015 年开始运行追赶。对于每个执行日期，任务实例在一分钟内完成。但是，第二天的任务仅在 5 分钟窗口内开始。例如。上午 10:00、上午 10:05、上午 10:10
当我在 WatchKit 中推送一个新 Controller 并在新 Controller 的awakeWithContext: 方法中使用 setTitle 时，它需要一秒钟左右来设置标题，直到
我将图像显示为 SVG 文件和文本。 出于某种原因，svg 图像的渲染速度比屏幕的其余部分慢，从而导致延迟，这对用户体验不利。 这种延迟正常吗？我该怎么做才能让整个屏幕同时呈现？ Row( ma
我正在考虑在我的应用程序中使用 firebase 动态链接。我需要将唯一标识符从电子邮件生成的链接传递到用户应用程序中。当用户安装了应用程序时，这可以正常工作，但是，我对未安装应用程序的方式有些困惑。
您知道如何使用 JQuery 的延迟方法和一个函数来检测所有已更改的表单并将每个表单作为 Ajax 帖子提交吗？ 如果我只列出大量表单提交，我可以得到同样的结果，但如果我使用... $('form.c
我需要一种方法来通过回调获取不同的脚本。这个方法工作正常: fetchScripts:function() { var _this=this; $.when( $.aj
我编写了一个 jquery 脚本，允许我淡入和淡出 div，然后重复。该代码运行良好。但是，当我尝试添加延迟(我希望 div 在淡出之前保持几秒钟)时，它无法正常工作。我尝试在代码中的几个地方添加延迟
我正在努力在延迟、带宽和吞吐量之间划清界限。 有人可以用简单的术语和简单的例子来解释我吗？ 最佳答案 水比喻: 延迟 是穿过管子所需的时间。 带宽是管有多宽。 水流量为吞吐量 车辆类比: 从源到目的地
我有一个 CRM 系统，当添加联系人时，我想将他们添加到会计系统中。 我在 CRM 系统中设置了一个 Webhook，将联系人传递给 Azure 函数。 Azure 函数连接到会计系统 API 并在那
我有一个 Android AudioTrack，例如: private AudioTrack mAudioTrack; int min = AudioTrack.getMinBufferSize(sa
我正在 React 中开发一个 TODO 应用程序，并尝试构建将删除选中项目延迟 X 秒的功能，并且如果在这段时间内未选中该框，它将不会被删除。 我遇到的主要问题是当用户在同一 X 秒内检查、取消检查
我是一名优秀的程序员,十分优秀！