第Golang分布式應(yīng)用定時(shí)任務(wù)示例詳解目錄正文最小堆時(shí)間輪總結(jié)

正文

在系統(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()

returnnil

//核心邏輯

func(c*Cron)runTask(){

now:=time.Now()

duration:=infTime

//獲取堆頂元素

task,ok:=c.tasks.Peek()

ifok{

//如果已刪除則彈出

if!c.set.Has(task.Name()){

c.tasks.Pop()

return

//計(jì)算于當(dāng)前時(shí)間查找，設(shè)置定時(shí)器

iftask.next.After(now){

duration=task.next.Sub(now)

}else{

duration=0

timer:=time.NewTimer(duration)

defertimer.Stop()

//當(dāng)有新元素插入直接返回，防止新元素執(zhí)行時(shí)間小于當(dāng)前堆頂元素

select{

case-c.new:

return

case-timer.C:

//彈出任務(wù)，執(zhí)行

gotask.Exec()

//計(jì)算下次執(zhí)行時(shí)間，如果為0說明任務(wù)已結(jié)束，否則重新入堆

task.next=task.Next(time.Now())

iftask.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í)間輪插入刪除元素更快。

核心代碼如下:

//定義

typeTimeWheelstruct{

intervaltime.Duration//觸發(fā)間隔

slotsint//總插槽數(shù)

currentSlotint//當(dāng)前插槽數(shù)

tasks[]*list.List//環(huán)形列表，每個(gè)元素為對(duì)應(yīng)插槽的任務(wù)列表

setcontainerx.Set[string]//記錄所有任務(wù)key值，用來檢查任務(wù)是否被刪除

tricker*time.Ticker//定時(shí)觸發(fā)器

loggerlogr.Logger

func(tw*TimeWheel)Run()error{

tw.tricker=time.NewTicker(erval)

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

returnnil

func(tw*TimeWheel)RunTask(nowtime.Time,slotint){

//一次執(zhí)行任務(wù)列表

foritem:=taskList.Front();item!=nil;{

task,ok:=item.Value.(*TimeWheelTask)

//任務(wù)圈數(shù)大于0，不需要執(zhí)行，將圈數(shù)減一

iftask.circle0{

task.circle--

item=item.Next()

continue

//運(yùn)行任務(wù)

gotask.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(nowtime.Time,task*TimeWheelTask){

if!task.initialized{

task.next=task.Next(now)

task.initialized=true

duration:=task.next.Sub(now)

ifduration=0{

task.slot=tw.currentSlot+1

task.circle=0

}else{

mult:=int(duration/erval)

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ù)按照一定算