Golang如何實(shí)現(xiàn)CronJob

這篇文章主要講解了“Golang如何實(shí)現(xiàn)CronJob”，文中的講解內(nèi)容簡(jiǎn)單清晰，易于學(xué)習(xí)與理解，下面請(qǐng)大家跟著小編的思路慢慢深入，一起來(lái)研究和學(xué)習(xí)“Golang如何實(shí)現(xiàn)CronJob”吧！

引言

最近做了一個(gè)需求，是定時(shí)任務(wù)相關(guān)的。以前定時(shí)任務(wù)都是通過(guò) linux crontab 去實(shí)現(xiàn)的，現(xiàn)在服務(wù)上云(k8s)了，嘗試了 k8s 的 CronJob，由于公司提供的是界面化工具，使用、查看起來(lái)很不方便。于是有了本文，通過(guò)一個(gè)單 pod 去實(shí)現(xiàn)一個(gè)常駐服務(wù)，去跑定時(shí)任務(wù)。

經(jīng)過(guò)篩選，選用了cron這個(gè)庫(kù)，它支持 linux cronjob 語(yǔ)法取配置定時(shí)任務(wù)，還支持@every 10s、@hourly 等描述符去配置定時(shí)任務(wù)，完全滿足我們要求，比如下面的例子：

package main

import (
	"fmt"

	"github.com/natefinch/lumberjack"
	"github.com/robfig/cron/v3"
	"github.com/sirupsen/logrus"
)

type CronLogger struct {
	clog *logrus.Logger
}

func (l *CronLogger) Info(msg string, keysAndValues ...interface{}) {
	l.clog.WithFields(logrus.Fields{
		"data": keysAndValues,
	}).Info(msg)
}

func (l *CronLogger) Error(err error, msg string, keysAndValues ...interface{}) {
	l.clog.WithFields(logrus.Fields{
		"msg":  msg,
		"data": keysAndValues,
	}).Warn(err.Error())
}

func main() {
	logger := logrus.New()
	_logger := &lumberjack.Logger{
		Filename:   "./test.log",
		MaxSize:    50,
		MaxAge:     15,
		MaxBackups: 5,
	}

	logger.SetOutput(_logger)
	logger.SetFormatter(&logrus.JSONFormatter{
		DisableHTMLEscape: true,
	})

	c := cron.New(cron.WithLogger(&CronLogger{
		clog: logger,
	}))
	c.AddFunc("*/5 * * * *", func() {
		fmt.Println("你的流量包即將過(guò)期了")
	})
	c.AddFunc("*/2 * * * *", func() {
		fmt.Println("你的轉(zhuǎn)碼包即將過(guò)期了")
	})
	c.Start()

	for {
		select {}
	}
}

使用了 cronjob、并結(jié)合了 golang 的 log 組建，輸出日志到文件，使用很方便。

但是，在使用過(guò)程中，發(fā)現(xiàn)還有些不足，缺少某些功能，比如我很想使用的查看任務(wù)列表。

類庫(kù)介紹

擴(kuò)展性強(qiáng)

此類庫(kù)擴(kuò)展性挺強(qiáng)，通過(guò) JobWrapper 去包裝一個(gè)任務(wù)，NewChain(w1, w2, w3).Then(job)，相關(guān)實(shí)現(xiàn)如下：

type JobWrapper func(Job) Job
type Chain struct {
    wrappers []JobWrapper
}
func NewChain(c ...JobWrapper) Chain {
    return Chain{c}
}
func (c Chain) Then(j Job) Job {
    for i := range c.wrappers {
        j = c.wrappers[len(c.wrappers)-i-1](j)
    }
    return j
}

比如當(dāng)前腳本如果還沒有執(zhí)行完，下次任務(wù)時(shí)間又到了，就可以通過(guò)如下默認(rèn)提供的 wrapper 去避免繼續(xù)執(zhí)行?？梢钥吹阶詈髨?zhí)行的任務(wù) j.Run() 被包裝在了一個(gè)函數(shù)閉包中，并且根據(jù)閉包中的 channel 去判斷是否執(zhí)行，避免重復(fù)執(zhí)行。首次執(zhí)行的時(shí)候，容量為 1 的 channel 中已經(jīng)有數(shù)據(jù)了，重復(fù)執(zhí)行時(shí)，channel 無(wú)數(shù)據(jù)，默認(rèn)跳過(guò)，等上次任務(wù)執(zhí)行完成后，又像 channel 中寫入一條數(shù)據(jù)，下次 channel 可以讀出數(shù)據(jù)，又可以執(zhí)行任務(wù)了：

func SkipIfStillRunning(j Job) Job {
    var ch = make(chan struct{}, 1)
    ch <- struct{}{}
    return FuncJob(func() {
        select {
        case v := <-ch:
            defer func() { ch <- v }()
            j.Run()
        default:
            // "skip"
        }
    })
}

主流程

cron 主流程是啟動(dòng)一個(gè)協(xié)程，里面有雙重 for 循環(huán)，下面我們來(lái)一起分析一下。

定時(shí)器

第一層循環(huán)，首先計(jì)算下次最早執(zhí)行任務(wù)的時(shí)間跟當(dāng)前時(shí)間間隔 gap，然后設(shè)置定時(shí)器為 gap，這里很巧妙，定時(shí)器間隔不是 1s/次，而是跟下次任務(wù)的時(shí)間相關(guān)，這樣就避免了無(wú)用的定時(shí)器循環(huán)，也讓執(zhí)行時(shí)間更精準(zhǔn)，不存在設(shè)置小了浪費(fèi)資源，設(shè)置大了誤差大的情況。接下來(lái)進(jìn)入第二層循環(huán)。

sort.Sort(byTime(c.entries))
timer = time.NewTimer(c.entries[0].Next.Sub(now))

事件循環(huán)

事件循環(huán)中，包含了很多事件，比如 添加任務(wù)、停止、移除任務(wù)，當(dāng) cron 啟動(dòng)之后，這些任務(wù)都是異步的。比如添加任務(wù)，不會(huì)直接將任務(wù)信息寫入內(nèi)存中，而是進(jìn)入事件循環(huán)，加入之后，重新計(jì)算第一二層循環(huán)，避免了正在修改任務(wù)信息，又執(zhí)行任務(wù)信息，然后出錯(cuò)的情況。

有人可能會(huì)問(wèn)了，為何不在事件中加鎖，這樣也能避免內(nèi)存競(jìng)爭(zhēng)。我想說(shuō)，我們執(zhí)行的是腳本任務(wù)，有的事件可能很長(zhǎng)，可能會(huì)阻塞有些事件，所以這些事件都放在循環(huán)中，避免了加鎖，也滿足了要求。

for {
    select {
    case now = <-timer.C:
        // 執(zhí)行任務(wù)
    case newEntry := <-c.add:
        // 添加任務(wù)
    case replyChan := <-c.snapshot:
        // 獲取任務(wù)信息
    case <-c.stop:
        //  停止任務(wù)
    case id := <-c.remove:
        // 移除任務(wù)
    }
    break
}

類庫(kù)改造

在了解了項(xiàng)目的基本情況之后，對(duì)項(xiàng)目做了部分改造，方便使用。

打印任務(wù)列表信息

在主循環(huán)匯總加入了信號(hào)量監(jiān)聽，當(dāng)觸發(fā)信號(hào)量 SIGUSR1，將任務(wù)信息輸出到日志：

usrSig := make(chan os.Signal, 1)
signal.Notify(usrSig, syscall.SIGUSR1)

for {
	select {
	case <-usrSig:
		// 啟動(dòng)單獨(dú)的協(xié)程去打印定時(shí)任務(wù)執(zhí)行信息
		continue
	}
	break
}

根據(jù)名稱移除腳本

目前腳本只能根據(jù)腳本 id 去移除要執(zhí)行的任務(wù)，執(zhí)行過(guò)程中，也不能通過(guò)命令去移除任務(wù)，不是太方便。比如有個(gè)腳本馬上要執(zhí)行了，但是該腳本發(fā)現(xiàn)問(wèn)題了，這時(shí)候生產(chǎn)環(huán)境的話，就需要更新代碼，然后重啟服務(wù)去下線腳本任務(wù)，這時(shí)候，黃花菜可能都涼了。

所以我也是通過(guò)信號(hào)量，來(lái)處理運(yùn)行之后，運(yùn)行中移除任務(wù)的問(wèn)題，收到信號(hào)量之后，讀取文件中的內(nèi)容，根據(jù)命令去處理 runing 中的內(nèi)存：

usrSig2 := make(chan os.Signal, 1)
signal.Notify(usrSig2, syscall.SIGUSR2)

......
case <-usrSig2:
	actionByte, err := os.ReadFile("/tmp/cron.action")
	...... //校驗(yàn)命令正確性
	action := strings.Fields(string(actionByte))
	switch action[0] {
	case "removeTag":
		timer.Stop()
		now = c.now()
		c.removeEntryByTag(action[1])
		c.logger.Info("removedByTag", "tag", action[1])
	}
......

改造效果

由于原項(xiàng)目已經(jīng) 2 年多沒有個(gè)更新過(guò)了，就算發(fā)起 pr 估計(jì)也不會(huì)被處理，所以 fork 一份放在了這里aizuyan/cron進(jìn)行改造，下面是改進(jìn)之后的代碼：

package main

import (
	// 加載配置文件

	"fmt"

	"github.com/aizuyan/cron/v3"
)

func main() {
	c := cron.New(cron.WithLogger(cron.DefaultLogger))
	c.AddFuncWithTag("流量包過(guò)期", "*/5 * * * *", func() {
		fmt.Println("你的流量包即將過(guò)期了")
	})
	c.AddFuncWithTag("轉(zhuǎn)碼包過(guò)期", "*/2 * * * *", func() {
		fmt.Println("你的轉(zhuǎn)碼包即將過(guò)期了")
	})
	c.Start()

	for {
		select {}
	}
}

對(duì)每個(gè)定時(shí)任務(wù)增加了一個(gè)名稱標(biāo)識(shí)，當(dāng)任務(wù)啟動(dòng)后，當(dāng)我們執(zhí)行 kill -SIGUSR1 <pid> 的時(shí)候，會(huì)看到 stdout 輸出了運(yùn)行的任務(wù)列表信息：

+----+------------+-------------+---------------------+---------------------+
| ID |    TAG     |    SPEC     |        PREV         |        NEXT         |
+----+------------+-------------+---------------------+---------------------+
|  2 | 轉(zhuǎn)碼包過(guò)期 | */2 * * * * | 0001-01-01 00:00:00 | 2023-04-02 17:22:00 |
|  1 | 流量包過(guò)期 | */5 * * * * | 0001-01-01 00:00:00 | 2023-04-02 17:25:00 |
+----+------------+-------------+---------------------+---------------------+

執(zhí)行 kill -SIGUSR2 <pid>，移除轉(zhuǎn)碼包過(guò)期任務(wù)，避免了使用 ID 容易出錯(cuò)的問(wèn)題。

cat /tmp/cron.action 
removeTag 轉(zhuǎn)碼包過(guò)期
// {"data":["tag","轉(zhuǎn)碼包過(guò)期"],"level":"info","msg":"removedByTag","time":"2023-04-02T18:32:56+08:00"}

感謝各位的閱讀，以上就是“Golang如何實(shí)現(xiàn)CronJob”的內(nèi)容了，經(jīng)過(guò)本文的學(xué)習(xí)后，相信大家對(duì)Golang如何實(shí)現(xiàn)CronJob這一問(wèn)題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是億速云，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！