Golang分布式應(yīng)用定時(shí)任務(wù)示例詳解
正文
在系統(tǒng)開發(fā)中,有一類任務(wù)不是立即執(zhí)行,而是在未來某個(gè)時(shí)間點(diǎn)或者按照一定間隔去執(zhí)行,比如日志定期壓縮、報(bào)表制作、過期數(shù)據(jù)清理等,這就是定時(shí)任務(wù)。
在單機(jī)中,定時(shí)任務(wù)通常需要實(shí)現(xiàn)一個(gè)類似crontab的系統(tǒng),一般有兩種方式:
- 最小堆,按照任務(wù)執(zhí)行時(shí)間建堆,每次取最近的任務(wù)執(zhí)行
- 時(shí)間輪,將任務(wù)放到時(shí)間輪列表中,每次轉(zhuǎn)動(dòng)取對(duì)應(yīng)的任務(wù)列表執(zhí)行
最小堆
最小堆是一種特殊的完全二叉樹,任意非葉子節(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已啟動(dòng),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)動(dòng)時(shí)間輪,取插槽中任務(wù)列表執(zhí)行,如圖所示:

時(shí)間輪可看作一個(gè)表盤,如圖中時(shí)間間隔為1秒,總共60個(gè)格子,如果任務(wù)在3秒后執(zhí)行則放為插槽3,每秒轉(zhuǎn)動(dòng)次取插槽上所有任務(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)。對(duì)比最小堆,時(shí)間輪插入刪除元素更快。
核心代碼如下:
// 定義
type TimeWheel struct {
interval time.Duration // 觸發(fā)間隔
slots int // 總插槽數(shù)
currentSlot int // 當(dāng)前插槽數(shù)
tasks []*list.List // 環(huán)形列表,每個(gè)元素為對(duì)應(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)動(dòng)
now, ok := <-tw.tricker.C
if !ok {
break
}
// 轉(zhuǎn)動(dòng)一次,執(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ù)存在對(duì)應(yīng)插槽的時(shí)間
- 通過定時(shí)間模擬時(shí)間輪轉(zhuǎn)動(dòng)
- 每次到期后遍歷當(dāng)前插槽的任務(wù)列表,若任務(wù)圈數(shù)為0則執(zhí)行
- 如果任務(wù)未結(jié)束,計(jì)算下次執(zhí)行的插槽與圈數(shù)
- 轉(zhuǎn)入步驟2,依次執(zhí)行
總結(jié)
本文主要總結(jié)了定時(shí)任務(wù)的兩種實(shí)現(xiàn)方式,最小堆與時(shí)間輪,并分析其核心實(shí)現(xiàn)邏輯。
對(duì)于執(zhí)行分布式定時(shí)任務(wù),可以借助延時(shí)消息隊(duì)列或者直接使用Kubernetes的CronJob。
自己開發(fā)的話可以借助Etcd:
- 中心節(jié)點(diǎn)Coordinator將任務(wù)按照一定算法(Hash、輪詢、或者更復(fù)雜的分配算法)將任務(wù)與工作節(jié)點(diǎn)Worker綁定
- 每個(gè)Worker添加到有綁定到自己的任務(wù)則取出放到本地的Cron中
- 如果Worker掛掉,執(zhí)行將其上任務(wù)重新綁定即可
本文所有代碼見github.com/qingwave/go…
以上就是Golang分布式應(yīng)用定時(shí)任務(wù)示例詳解的詳細(xì)內(nèi)容,更多關(guān)于Golang分布式定時(shí)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章!
相關(guān)文章
Golang限流器time/rate設(shè)計(jì)與實(shí)現(xiàn)詳解
在?Golang?庫中官方給我們提供了限流器的實(shí)現(xiàn)golang.org/x/time/rate,它是基于令牌桶算法(Token?Bucket)設(shè)計(jì)實(shí)現(xiàn)的,下面我們就來看看他的具體使用吧2024-03-03
GoLang職責(zé)鏈模式代碼實(shí)現(xiàn)介紹
這篇文章主要介紹了GoLang職責(zé)鏈模式代碼實(shí)現(xiàn),職責(zé)鏈模式是一種常用的設(shè)計(jì)模式,可以提高代碼的靈活性與可維護(hù)性,職責(zé)鏈模式將請(qǐng)求和處理分離,可以讓請(qǐng)求在處理鏈中依次經(jīng)過多個(gè)處理者,直到找到能夠處理請(qǐng)求的處理者為止2023-05-05
go?micro微服務(wù)proto開發(fā)安裝及使用規(guī)則
這篇文章主要為大家介紹了go?micro微服務(wù)proto開發(fā)中安裝Protobuf及基本規(guī)范字段的規(guī)則詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-01-01
Golang?Redis連接池實(shí)現(xiàn)原理及示例探究
這篇文章主要為大家介紹了Golang?Redis連接池實(shí)現(xiàn)示例探究,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2024-01-01
一文詳解Golang?定時(shí)任務(wù)庫?gron?設(shè)計(jì)和原理
這篇文章主要介紹了一文詳解Golang?定時(shí)任務(wù)庫?gron?設(shè)計(jì)和原理,gron是一個(gè)比較小巧、靈活的定時(shí)任務(wù)庫,可以執(zhí)行定時(shí)的、周期性的任務(wù)。gron提供簡(jiǎn)潔的、并發(fā)安全的接口2022-08-08
Golang并發(fā)繞不開的重要組件之Channel詳解
Channel是一個(gè)提供可接收和發(fā)送特定類型值的用于并發(fā)函數(shù)通信的數(shù)據(jù)類型,也是Golang并發(fā)繞不開的重要組件之一,本文就來和大家深入聊聊Channel的相關(guān)知識(shí)吧2023-06-06
Go基本數(shù)據(jù)類型與string類型互轉(zhuǎn)
本文主要介紹了Go基本數(shù)據(jù)類型與string類型互轉(zhuǎn),文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-03-03

