網(wǎng)上有很多關(guān)于pos機延遲到賬,zero 如何應對海量定時(shí)的知識，也有很多人為大家解答關(guān)于pos機延遲到賬的問(wèn)題，今天pos機之家(www.xjcwpx.cn)為大家整理了關(guān)于這方面的知識，讓我們一起來(lái)看下吧!
本文目錄一覽：
1、pos機延遲到賬
pos機延遲到賬
一個(gè)系統中存在著(zhù)大量的調度任務(wù)，同時(shí)調度任務(wù)存在時(shí)間的滯后性，而大量的調度任務(wù)如果每一個(gè)都使用自己的調度器來(lái)管理任務(wù)的生命周期的話(huà)，浪費cpu的資源而且很低效。
本文來(lái)介紹 go-zero 中 延遲操作，它可能讓開(kāi)發(fā)者調度多個(gè)任務(wù)時(shí)，只需關(guān)注具體的業(yè)務(wù)執行函數和執行時(shí)間「立即或者延遲」。而 延遲操作，通?？梢圆捎脙蓚€(gè)方案：
Timer：定時(shí)器維護一個(gè)優(yōu)先隊列，到時(shí)間點(diǎn)執行，然后把需要執行的 task 存儲在 map 中collection 中的 timingWheel ，維護一個(gè)存放任務(wù)組的數組，每一個(gè)槽都維護一個(gè)存儲task的雙向鏈表。開(kāi)始執行時(shí)，計時(shí)器每隔指定時(shí)間執行一個(gè)槽里面的tasks。方案2把維護task從 優(yōu)先隊列 O(nlog(n)) 降到 雙向鏈表 O(1)，而執行task也只要輪詢(xún)一個(gè)時(shí)間點(diǎn)的tasks O(N)，不需要像優(yōu)先隊列，放入和刪除元素 O(nlog(n))。
我們先看看 go-zero 中自己對 timingWheel 的使用 ：
cache 中的 timingWheel首先我們先來(lái)在 collection 的 cache 中關(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ò)期處理，參數依次代表：
interval：時(shí)間劃分刻度numSlots：時(shí)間槽execute：時(shí)間點(diǎn)執行函數在 cache 中執行函數則是 刪除過(guò)期key，而這個(gè)過(guò)期則由 timingWheel 來(lái)控制推進(jìn)時(shí)間。
接下來(lái)，就通過(guò) cache 對 timingWheel 的使用來(lái)認識。
初始化// 真正做初始化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(), // 存儲task{key, value}的map [執行execute所需要的參數] tickedPos: numSlots - 1, // at previous virtual circle execute: execute, // 執行函數 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 中存儲的 list 全部準備好 tw.initSlots() // 開(kāi)啟異步協(xié)程，使用 channel 來(lái)做task通信和傳遞 go tw.run() return tw, nil}
以上比較直觀(guān)展示 timingWheel 的 “時(shí)間輪”，后面會(huì )圍繞這張圖解釋其中推進(jìn)的細節。
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 執行，并以internal時(shí)間段轉動(dòng)，然后底層不停的獲取來(lái)自 slot 中的 list 的task，交給 execute 執行。
Task Operation緊接著(zhù)就是設置 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()第一次設置key -> SetTimer()
所以對于 timingWheel 的使用上就清晰了，開(kāi)發(fā)者根據需求可以 add 或是 update。
同時(shí)我們跟源碼進(jìn)去會(huì )發(fā)現：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()
由上面的調用鏈，有一個(gè)都會(huì )調用的函數：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ù)應該立即執行 if task.delay < tw.interval { threading.GoSafe(func() { tw.execute(timer.item.key, timer.item.value) }) return } // 如果 > interval，則通過(guò) 延遲時(shí)間delay 計算其出時(shí)間輪中的 new pos, circle pos, circle := tw.getPositionAndCircle(task.delay) if pos >= timer.pos { timer.item.circle = circle // 記錄前后的移動(dòng)offset。為了后面過(guò)程重新入隊 timer.item.diff = pos - timer.pos } else if circle > 0 { // 轉移到下一層，將 circle 轉換為 diff 一部分 circle-- timer.item.circle = circle // 因為是一個(gè)數組，要加上 numSlots [也就是相當于要走到下一層] timer.item.diff = tw.numSlots + pos - timer.pos } else { // 如果 offset 提前了，此時(shí) task 也還在第一層 // 標記刪除老的 task，并重新入隊，等待被執行 timer.item.removed = true newItem := &timingEntry{ baseEntry: task, value: timer.item.value, } tw.slots[pos].PushBack(newItem) tw.setTimerPosition(pos, newItem) }}
以上過(guò)程有以下幾種情況：
delay < internal：因為 < 單個(gè)時(shí)間精度，表示這個(gè)任務(wù)已經(jīng)過(guò)期，需要馬上執行針對改變的 delay：new >= old：<newPos, newCircle, diff>newCircle > 0：計算diff，并將 circle 轉換為 下一層，故diff + numslots如果只是單純延遲時(shí)間縮短，則將老的task標記刪除，重新加入list，等待下一輪loop被executeExecute之前在初始化中，run() 中定時(shí)器的不斷推進(jìn)，推進(jìn)的過(guò)程主要就是把 list中的 task 傳給執行的 execute func。我們從定時(shí)器的執行開(kāi)始看：
// 定時(shí)器 「每隔 internal 會(huì )執行一次」func (tw *TimingWheel) onTick() { // 每次執行更新一下當前執行 tick 位置 tw.tickedPos = (tw.tickedPos + 1) % tw.numSlots // 獲取此時(shí) tick位置 中的存儲task的雙向鏈表 l := tw.slots[tw.tickedPos] tw.scanAndRunTasks(l)}
緊接著(zhù)是如何去執行 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 { // 當前執行點(diǎn)已經(jīng)過(guò)期，但是同時(shí)不在第一層，所以當前層即然已經(jīng)完成了，就會(huì )降到下一層 // 但是并沒(méi)有修改 pos task.circle-- e = e.Next() continue } else if task.diff > 0 { // 因為之前已經(jīng)標注了diff，需要再進(jìn)入隊列 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 } // 以上的情況都是不能執行的情況，能夠執行的會(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 執行即可 tw.runTasks(tasks)}
具體的分支情況在注釋中說(shuō)明了，在看的時(shí)候可以和前面的 moveTask() 結合起來(lái)，其中 circle 下降，diff 的計算是關(guān)聯(lián)兩個(gè)函數的重點(diǎn)。
至于 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}
上面的過(guò)程可以簡(jiǎn)化成下面：
steps = d / intervalpos = step % numSlots - 1circle = (step - 1) / numSlots總結
timingWheel 靠定時(shí)器推動(dòng)，時(shí)間前進(jìn)的同時(shí)會(huì )取出當前時(shí)間格中 list「雙向鏈表」的task，傳遞到 execute 中執行。因為是是靠 internal 固定時(shí)間刻度推進(jìn)，可能就會(huì )出現：一個(gè) 60s 的task，internal = 1s，這樣就會(huì )空跑59次loop。
而在擴展時(shí)間上，采取 circle 分層，這樣就可以不斷復用原有的 numSlots ，因為定時(shí)器在不斷 loop，而執行可以把上層的 slot 下降到下層，在不斷 loop 中就可以執行到上層的task。這樣的設計可以在不創(chuàng )造額外的數據結構，突破長(cháng)時(shí)間的限制。
同時(shí)在 go-zero 中還有很多實(shí)用的組件工具，用好工具對于提升服務(wù)性能和開(kāi)發(fā)效率都有很大的幫助，希望本篇文章能給大家帶來(lái)一些收獲。
項目地址https://github.com/tal-tech/go-zero
好未來(lái)技術(shù)
以上就是關(guān)于pos機延遲到賬,zero 如何應對海量定時(shí)的知識，后面我們會(huì )繼續為大家整理關(guān)于pos機延遲到賬的知識，希望能夠幫助到大家！
