主頁(yè) > 知識(shí)庫(kù) > go-zero 應(yīng)對(duì)海量定時(shí)/延遲任務(wù)的技巧

go-zero 應(yīng)對(duì)海量定時(shí)/延遲任務(wù)的技巧

熱門(mén)標(biāo)簽:利用地圖標(biāo)注位置 赤峰電銷(xiāo) 江蘇呼叫中心外呼系統(tǒng)有效果嗎 杭州人工智能電銷(xiāo)機(jī)器人費(fèi)用 地圖區(qū)域圖標(biāo)注后導(dǎo)出 貴州電話(huà)智能外呼系統(tǒng) 官渡電銷(xiāo)外呼管理系統(tǒng)怎么收費(fèi) 400開(kāi)頭電話(huà)怎樣申請(qǐng) 谷歌美發(fā)店地圖標(biāo)注

一個(gè)系統(tǒng)中存在著大量的調(diào)度任務(wù),同時(shí)調(diào)度任務(wù)存在時(shí)間的滯后性,而大量的調(diào)度任務(wù)如果每一個(gè)都使用自己的調(diào)度器來(lái)管理任務(wù)的生命周期的話(huà),浪費(fèi)cpu的資源而且很低效。

本文來(lái)介紹 go-zero 中 延遲操作,它可能讓開(kāi)發(fā)者調(diào)度多個(gè)任務(wù)時(shí),只需關(guān)注具體的業(yè)務(wù)執(zhí)行函數(shù)和執(zhí)行時(shí)間「立即或者延遲」。而 延遲操作,通??梢圆捎脙蓚€(gè)方案:

Timer:定時(shí)器維護(hù)一個(gè)優(yōu)先隊(duì)列,到時(shí)間點(diǎn)執(zhí)行,然后把需要執(zhí)行的 task 存儲(chǔ)在 map 中collection 中的 timingWheel ,維護(hù)一個(gè)存放任務(wù)組的數(shù)組,每一個(gè)槽都維護(hù)一個(gè)存儲(chǔ)task的雙向鏈表。開(kāi)始執(zhí)行時(shí),計(jì)時(shí)器每隔指定時(shí)間執(zhí)行一個(gè)槽里面的tasks。

方案2把維護(hù)task從 優(yōu)先隊(duì)列 O(nlog(n)) 降到 雙向鏈表 O(1),而執(zhí)行task也只要輪詢(xún)一個(gè)時(shí)間點(diǎn)的tasks O(N),不需要像優(yōu)先隊(duì)列,放入和刪除元素 O(nlog(n))

我們先看看 go-zero 中自己對(duì) timingWheel 的使用 :

cache 中的 timingWheel

首先我們先來(lái)在 collectioncache 中關(guān)于 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 初始化中也同時(shí)初始化 timingWheel 做key的過(guò)期處理,參數(shù)依次代表:

  • interval:時(shí)間劃分刻度
  • numSlots:時(shí)間槽
  • execute:時(shí)間點(diǎn)執(zhí)行函數(shù)

cache 中執(zhí)行函數(shù)則是 刪除過(guò)期key,而這個(gè)過(guò)期則由 timingWheel 來(lái)控制推進(jìn)時(shí)間。

接下來(lái),就通過(guò) cache 對(duì) timingWheel 的使用來(lái)認(rèn)識(shí)。

初始化

// 真正做初始化
func newTimingWheelWithClock(interval time.Duration, numSlots int, execute Execute, ticker timex.Ticker) (
	*TimingWheel, error) {
	tw := TimingWheel{
		interval:  interval,      // 單個(gè)時(shí)間格時(shí)間間隔
		ticker:  ticker,      // 定時(shí)器,做時(shí)間推動(dòng),以interval為單位推進(jìn)
		slots:   make([]*list.List, numSlots), // 時(shí)間輪
		timers:  NewSafeMap(),     // 存儲(chǔ)task{key, value}的map [執(zhí)行execute所需要的參數(shù)]
		tickedPos:  numSlots - 1,     // at previous virtual circle
		execute:  execute,      // 執(zhí)行函數(shù)
		numSlots:  numSlots,      // 初始化 slots num
		setChannel: make(chan timingEntry),  // 以下幾個(gè)channel是做task傳遞的
		moveChannel: make(chan baseEntry),
		removeChannel: make(chan interface{}),
		drainChannel: make(chan func(key, value interface{})),
		stopChannel: make(chan lang.PlaceholderType),
	}
	// 把 slot 中存儲(chǔ)的 list 全部準(zhǔn)備好
	tw.initSlots()
	// 開(kāi)啟異步協(xié)程,使用 channel 來(lái)做task通信和傳遞
	go tw.run()

	return tw, nil
}

以上比較直觀(guān)展示 timingWheel 的 “時(shí)間輪”,后面會(huì)圍繞這張圖解釋其中推進(jìn)的細(xì)節(jié)。

go tw.run() 開(kāi)一個(gè)協(xié)程做時(shí)間推動(dòng):

func (tw *TimingWheel) run() {
	for {
		select {
  // 定時(shí)器做時(shí)間推動(dòng) -> scanAndRunTasks()
		case -tw.ticker.Chan():
			tw.onTick()
  // add task 會(huì)往 setChannel 輸入task
		case task := -tw.setChannel:
			tw.setTask(task)
		...
		}
	}
}

可以看出,在初始化的時(shí)候就開(kāi)始了 timer 執(zhí)行,并以internal時(shí)間段轉(zhuǎn)動(dòng),然后底層不停的獲取來(lái)自 slot 中的 list 的task,交給 execute 執(zhí)行。

Task Operation

緊接著就是設(shè)置 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 中有沒(méi)有存在這個(gè)key存在,則更新 expire -> MoveTimer()第一次設(shè)置key -> SetTimer()

所以對(duì)于 timingWheel 的使用上就清晰了,開(kāi)發(fā)者根據(jù)需求可以 add 或是 update

同時(shí)我們跟源碼進(jìn)去會(huì)發(fā)現(xiàn):SetTimer() MoveTimer() 都是將task輸送到channel,由 run() 中開(kāi)啟的協(xié)程不斷取出 channel 的task操作。

SetTimer() -> setTask()

not exist task:getPostion -> pushBack to list -> setPositionexist task:get from timers -> moveTask()

MoveTimer() -> moveTask()

由上面的調(diào)用鏈,有一個(gè)都會(huì)調(diào)用的函數(shù):moveTask()

func (tw *TimingWheel) moveTask(task baseEntry) {
	// timers: Map => 通過(guò)key獲取 [positionEntry「pos, task」]
	val, ok := tw.timers.Get(task.key)
	if !ok {
		return
	}

	timer := val.(*positionEntry)
 	// {delay  interval} => 延遲時(shí)間比一個(gè)時(shí)間格間隔還小,沒(méi)有更小的刻度,說(shuō)明任務(wù)應(yīng)該立即執(zhí)行
	if task.delay  tw.interval {
		threading.GoSafe(func() {
			tw.execute(timer.item.key, timer.item.value)
		})
		return
	}
	// 如果 > interval,則通過(guò) 延遲時(shí)間delay 計(jì)算其出時(shí)間輪中的 new pos, circle
	pos, circle := tw.getPositionAndCircle(task.delay)
	if pos >= timer.pos {
		timer.item.circle = circle
    // 記錄前后的移動(dòng)offset。為了后面過(guò)程重新入隊(duì)
		timer.item.diff = pos - timer.pos
	} else if circle > 0 {
		// 轉(zhuǎn)移到下一層,將 circle 轉(zhuǎn)換為 diff 一部分
		circle--
		timer.item.circle = circle
		// 因?yàn)槭且粋€(gè)數(shù)組,要加上 numSlots [也就是相當(dāng)于要走到下一層]
		timer.item.diff = tw.numSlots + pos - timer.pos
	} else {
		// 如果 offset 提前了,此時(shí) task 也還在第一層
		// 標(biāo)記刪除老的 task,并重新入隊(duì),等待被執(zhí)行
		timer.item.removed = true
		newItem := timingEntry{
			baseEntry: task,
			value:  timer.item.value,
		}
		tw.slots[pos].PushBack(newItem)
		tw.setTimerPosition(pos, newItem)
	}
}

以上過(guò)程有以下幾種情況:

delay internal:因?yàn)? 單個(gè)時(shí)間精度,表示這個(gè)任務(wù)已經(jīng)過(guò)期,需要馬上執(zhí)行針對(duì)改變的 delaynew >= oldnewPos, newCircle, diff>newCircle > 0:計(jì)算diff,并將 circle 轉(zhuǎn)換為 下一層,故diff + numslots如果只是單純延遲時(shí)間縮短,則將老的task標(biāo)記刪除,重新加入list,等待下一輪loop被execute

Execute

之前在初始化中,run() 中定時(shí)器的不斷推進(jìn),推進(jìn)的過(guò)程主要就是把 list中的 task 傳給執(zhí)行的 execute func。我們從定時(shí)器的執(zhí)行開(kāi)始看:

// 定時(shí)器 「每隔 internal 會(huì)執(zhí)行一次」
func (tw *TimingWheel) onTick() {
  // 每次執(zhí)行更新一下當(dāng)前執(zhí)行 tick 位置
	tw.tickedPos = (tw.tickedPos + 1) % tw.numSlots
  // 獲取此時(shí) tick位置 中的存儲(chǔ)task的雙向鏈表
	l := tw.slots[tw.tickedPos]
	tw.scanAndRunTasks(l)
}

緊接著是如何去執(zhí)行 execute

func (tw *TimingWheel) scanAndRunTasks(l *list.List) {
	// 存儲(chǔ)目前需要執(zhí)行的task{key, value} [execute所需要的參數(shù),依次傳遞給execute執(zhí)行]
	var tasks []timingTask

	for e := l.Front(); e != nil; {
		task := e.Value.(*timingEntry)
    // 標(biāo)記刪除,在 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 {
			// 當(dāng)前執(zhí)行點(diǎn)已經(jīng)過(guò)期,但是同時(shí)不在第一層,所以當(dāng)前層即然已經(jīng)完成了,就會(huì)降到下一層
      // 但是并沒(méi)有修改 pos
			task.circle--
			e = e.Next()
			continue
		} else if task.diff > 0 {
			// 因?yàn)橹耙呀?jīng)標(biāo)注了diff,需要再進(jìn)入隊(duì)列
			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
		}
		// 以上的情況都是不能執(zhí)行的情況,能夠執(zhí)行的會(huì)被加入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,然后把每個(gè) task->execute 執(zhí)行即可
	tw.runTasks(tasks)
}

具體的分支情況在注釋中說(shuō)明了,在看的時(shí)候可以和前面的 moveTask() 結(jié)合起來(lái),其中 circle 下降,diff 的計(jì)算是關(guān)聯(lián)兩個(gè)函數(shù)的重點(diǎn)。

至于 diff 計(jì)算就涉及到 pos, circle 的計(jì)算:

// 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
}

上面的過(guò)程可以簡(jiǎn)化成下面:

steps = d / interval
pos = step % numSlots - 1
circle = (step - 1) / numSlots

總結(jié)

timingWheel 靠定時(shí)器推動(dòng),時(shí)間前進(jìn)的同時(shí)會(huì)取出當(dāng)前時(shí)間格中 list「雙向鏈表」的task,傳遞到 execute 中執(zhí)行。因?yàn)槭鞘强?internal 固定時(shí)間刻度推進(jìn),可能就會(huì)出現(xiàn):一個(gè) 60s 的task,internal = 1s,這樣就會(huì)空跑59次loop。

而在擴(kuò)展時(shí)間上,采取 circle 分層,這樣就可以不斷復(fù)用原有的 numSlots ,因?yàn)槎〞r(shí)器在不斷 loop,而執(zhí)行可以把上層的 slot 下降到下層,在不斷 loop 中就可以執(zhí)行到上層的task。這樣的設(shè)計(jì)可以在不創(chuàng)造額外的數(shù)據(jù)結(jié)構(gòu),突破長(zhǎng)時(shí)間的限制。

同時(shí)在 go-zero 中還有很多實(shí)用的組件工具,用好工具對(duì)于提升服務(wù)性能和開(kāi)發(fā)效率都有很大的幫助,希望本篇文章能給大家?guī)?lái)一些收獲。

項(xiàng)目地址

https://github.com/tal-tech/go-zero

好未來(lái)技術(shù)

到此這篇關(guān)于go-zero 如何應(yīng)對(duì)海量定時(shí)/延遲任務(wù)的文章就介紹到這了,更多相關(guān)go-zero定時(shí)/延遲任務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

您可能感興趣的文章:
  • 利用go-zero在Go中快速實(shí)現(xiàn)JWT認(rèn)證的步驟詳解
  • 如何用go-zero 實(shí)現(xiàn)中臺(tái)系統(tǒng)

標(biāo)簽:鷹潭 松原 宜春 武漢 泰安 黔西 保定 河池

巨人網(wǎng)絡(luò)通訊聲明:本文標(biāo)題《go-zero 應(yīng)對(duì)海量定時(shí)/延遲任務(wù)的技巧》,本文關(guān)鍵詞  go-zero,應(yīng)對(duì),海量,定時(shí),延遲,;如發(fā)現(xiàn)本文內(nèi)容存在版權(quán)問(wèn)題,煩請(qǐng)?zhí)峁┫嚓P(guān)信息告之我們,我們將及時(shí)溝通與處理。本站內(nèi)容系統(tǒng)采集于網(wǎng)絡(luò),涉及言論、版權(quán)與本站無(wú)關(guān)。
  • 相關(guān)文章
  • 下面列出與本文章《go-zero 應(yīng)對(duì)海量定時(shí)/延遲任務(wù)的技巧》相關(guān)的同類(lèi)信息!
  • 本頁(yè)收集關(guān)于go-zero 應(yīng)對(duì)海量定時(shí)/延遲任務(wù)的技巧的相關(guān)信息資訊供網(wǎng)民參考!
  • 推薦文章