成人六区,好大灬好硬灬好爽灬无码软件,五月天婷婷国产,欧美一区二区三色欲区AA大片,夜噜噜在线,欧洲在线,中文字幕,成人免费网站黄,国产黄在线观看

一個(gè)系統(tǒng)中存在著大量的調(diào)度任務(wù)，同時(shí)調(diào)度任務(wù)存在時(shí)間的滯后性，而大量的調(diào)度任務(wù)如果每一個(gè)都使用自己的調(diào)度器來(lái)管理任務(wù)的生命周期的話，浪費(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 也只要輪詢一個(gè)時(shí)間點(diǎn)的 tasks?O(N)，不需要像優(yōu)先隊(duì)列，放入和刪除元素?O(nlog(n))。

我們先看看?go-zero?中自己對(duì)?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ò)期處理，參數(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
}

以上比較直觀展示?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 -> setPosition
exist 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ì)改變的?delay：

new >= old：
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)一些收獲。

同時(shí)歡迎大家使用?go-zero?并加入我們，項(xiàng)目地址 https://github.com/tal-tech/go-zero

Go-Zero如何應(yīng)對(duì)海量定時(shí)/延遲任務(wù)

cache 中的 timingWheel

初始化

Task Operation

Execute

總結(jié)