Golang分布式應(yīng)用定時(shí)任務(wù)如何實(shí)現(xiàn)

這篇“Golang分布式應(yīng)用定時(shí)任務(wù)如何實(shí)現(xiàn)”文章的知識點(diǎn)大部分人都不太理解，所以小編給大家總結(jié)了以下內(nèi)容，內(nèi)容詳細(xì)，步驟清晰，具有一定的借鑒價(jià)值，希望大家閱讀完這篇文章能有所收獲，下面我們一起來看看這篇“Golang分布式應(yīng)用定時(shí)任務(wù)如何實(shí)現(xiàn)”文章吧。

最小堆

最小堆是一種特殊的完全二叉樹，任意非葉子節(jié)點(diǎn)的值不大于其子節(jié)點(diǎn)，如圖

通過最小堆，根據(jù)任務(wù)最近執(zhí)行時(shí)間鍵堆，每次取堆頂元素即最近需要執(zhí)行的任務(wù)，設(shè)置timer定時(shí)器，到期后觸發(fā)任務(wù)執(zhí)行。由于堆的特性每次調(diào)整的時(shí)間復(fù)雜度為O(lgN)，相較于普通隊(duì)列性能更快。

在container/heap中已經(jīng)實(shí)現(xiàn)操作堆的相關(guān)函數(shù)，我們只需要實(shí)現(xiàn)定期任務(wù)核心邏輯即可。

// 運(yùn)行
func (c *Cron) Run() error {
    // 設(shè)置cron已啟動，atomic.Bool來保證并發(fā)安全
	c.started.Store(true)
    // 主循環(huán)
	for {
        // 如果停止則退出
		if !c.started.Load() {
			break
		}
		c.runTask()
	}
	return nil
}
// 核心邏輯
func (c *Cron) runTask() {
	now := time.Now()
	duration := infTime
	// 獲取堆頂元素
	task, ok := c.tasks.Peek()
	if ok {
		// 如果已刪除則彈出
		if !c.set.Has(task.Name()) {
			c.tasks.Pop()
			return
		}
		// 計(jì)算于當(dāng)前時(shí)間查找，設(shè)置定時(shí)器
		if task.next.After(now) {
			duration = task.next.Sub(now)
		} else {
			duration = 0
		}
	}
	timer := time.NewTimer(duration)
	defer timer.Stop()
	// 當(dāng)有新元素插入直接返回，防止新元素執(zhí)行時(shí)間小于當(dāng)前堆頂元素
	select {
	case <-c.new:
		return
	case <-timer.C:
	}
	// 彈出任務(wù)，執(zhí)行
	go task.Exec()
	// 計(jì)算下次執(zhí)行時(shí)間，如果為0說明任務(wù)已結(jié)束，否則重新入堆
	task.next = task.Next(time.Now())
	if task.next.IsZero() {
		c.set.Delete(task.Name())
	} else {
		c.tasks.Push(task)
	}
}

主要邏輯可總結(jié)為:

• 將任務(wù)按照下次執(zhí)行時(shí)間建最小堆

• 每次取堆頂任務(wù)，設(shè)置定時(shí)器

• 如果中間有新加入任務(wù)，轉(zhuǎn)入步驟2

• 定時(shí)器到期后執(zhí)行任務(wù)

• 再次取下個(gè)任務(wù)，轉(zhuǎn)入步驟2，依次執(zhí)行

時(shí)間輪

另一種實(shí)現(xiàn)Cron的方式是時(shí)間輪，時(shí)間輪通過一個(gè)環(huán)形隊(duì)列，每個(gè)插槽放入需要到期執(zhí)行的任務(wù)，按照固定間隔轉(zhuǎn)動時(shí)間輪，取插槽中任務(wù)列表執(zhí)行，如圖所示:

時(shí)間輪可看作一個(gè)表盤，如圖中時(shí)間間隔為1秒，總共60個(gè)格子，如果任務(wù)在3秒后執(zhí)行則放為插槽3，每秒轉(zhuǎn)動次取插槽上所有任務(wù)執(zhí)行。

如果執(zhí)行時(shí)間超過最大插槽，比如有個(gè)任務(wù)需要63秒后執(zhí)行（超過了最大格子刻度），一般可以通過多層時(shí)間輪，或者設(shè)置一個(gè)額外變量圈數(shù)，只執(zhí)行圈數(shù)為0的任務(wù)。

時(shí)間輪插入的時(shí)間復(fù)雜度為O(1)，獲取任務(wù)列表復(fù)雜度為O(1)，執(zhí)行列表最差為O(n)。對比最小堆，時(shí)間輪插入刪除元素更快。

核心代碼如下:

// 定義
type TimeWheel struct {
	interval    time.Duration // 觸發(fā)間隔
	slots       int // 總插槽數(shù)
	currentSlot int // 當(dāng)前插槽數(shù)
	tasks       []*list.List // 環(huán)形列表，每個(gè)元素為對應(yīng)插槽的任務(wù)列表
	set         containerx.Set[string] // 記錄所有任務(wù)key值，用來檢查任務(wù)是否被刪除
	tricker *time.Ticker // 定時(shí)觸發(fā)器
	logger logr.Logger
}
func (tw *TimeWheel) Run() error {
	tw.tricker = time.NewTicker(tw.interval)
	for {
		// 通過定時(shí)器模擬時(shí)間輪轉(zhuǎn)動
		now, ok := <-tw.tricker.C
		if !ok {
			break
		}
		// 轉(zhuǎn)動一次，執(zhí)行任務(wù)列表
		tw.RunTask(now, tw.currentSlot)
		tw.currentSlot = (tw.currentSlot + 1) % tw.slots
	}
	return nil
}
func (tw *TimeWheel) RunTask(now time.Time, slot int) {
	// 一次執(zhí)行任務(wù)列表
	for item := taskList.Front(); item != nil; {
		task, ok := item.Value.(*TimeWheelTask)
		// 任務(wù)圈數(shù)大于0，不需要執(zhí)行，將圈數(shù)減一
		if task.circle > 0 {
			task.circle--
			item = item.Next()
			continue
		}
		// 運(yùn)行任務(wù)
		go task.Exec()
		// 計(jì)算任務(wù)下次運(yùn)行時(shí)間
		next := item.Next()
		taskList.Remove(item)
		item = next
		task.next = task.Next(now)
		if !task.next.IsZero() {
			tw.add(now, task)
		} else {
			tw.Remove(task.Name())
		}
	}
}
// 添加任務(wù)，計(jì)算下一次任務(wù)執(zhí)行的插槽與圈數(shù)
func (tw *TimeWheel) add(now time.Time, task *TimeWheelTask) {
	if !task.initialized {
		task.next = task.Next(now)
		task.initialized = true
	}
	duration := task.next.Sub(now)
	if duration <= 0 {
		task.slot = tw.currentSlot + 1
		task.circle = 0
	} else {
		mult := int(duration / tw.interval)
		task.slot = (tw.currentSlot + mult) % tw.slots
		task.circle = mult / tw.slots
	}
	tw.tasks[task.slot].PushBack(task)
	tw.set.Insert(task.Name())
}

時(shí)間輪的主要邏輯如下：

• 將任務(wù)存在對應(yīng)插槽的時(shí)間

• 通過定時(shí)間模擬時(shí)間輪轉(zhuǎn)動

• 每次到期后遍歷當(dāng)前插槽的任務(wù)列表，若任務(wù)圈數(shù)為0則執(zhí)行

• 如果任務(wù)未結(jié)束，計(jì)算下次執(zhí)行的插槽與圈數(shù)

• 轉(zhuǎn)入步驟2，依次執(zhí)行

以上就是關(guān)于“Golang分布式應(yīng)用定時(shí)任務(wù)如何實(shí)現(xiàn)”這篇文章的內(nèi)容，相信大家都有了一定的了解，希望小編分享的內(nèi)容對大家有幫助，若想了解更多相關(guān)的知識內(nèi)容，請關(guān)注億速云行業(yè)資訊頻道。