最近做了一個需求，是定時任務相關的。以前定時任務都是通過 linux crontab 去實現的，現在服務上雲(k8s)了，嘗試了 k8s 的 CronJob，由於公司提供的是介面化工具，使用、檢視起來很不方便。於是有了本文，通過一個單 pod 去實現一個常駐服務，去跑定時任務。
經過篩選，選用了 cron 這個庫，它支援 linux cronjob 語法取設定定時任務，還支援@every 10s、@hourly 等描述符去設定定時任務，完全滿足我們要求，比如下面的例子：
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("你的流量包即將過期了")
})
c.AddFunc("*/2 * * * *", func() {
fmt.Println("你的轉碼包即將過期了")
})
c.Start()
for {
select {}
}
}
使用了 cronjob、並結合了 golang 的 log 組建，輸出紀錄檔到檔案，使用很方便。
但是，在使用過程中，發現還有些不足，缺少某些功能，比如我很想使用的檢視任務列表。
此類庫擴充套件性挺強，通過 JobWrapper 去包裝一個任務，NewChain(w1, w2, w3).Then(job)，相關實現如下：
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
}
比如當前指令碼如果還沒有執行完，下次任務時間又到了，就可以通過如下預設提供的 wrapper 去避免繼續執行。可以看到最後執行的任務 j.Run() 被包裝在了一個函數閉包中，並且根據閉包中的 channel 去判斷是否執行，避免重複執行。首次執行的時候，容量為 1 的 channel 中已經有資料了，重複執行時，channel 無資料，預設跳過，等上次任務執行完成後，又像 channel 中寫入一條資料，下次 channel 可以讀出資料，又可以執行任務了：
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 主流程是啟動一個協程，裡面有雙重 for 迴圈，下面我們來一起分析一下。
第一層迴圈，首先計算下次最早執行任務的時間跟當前時間間隔 gap，然後設定定時器為 gap，這裡很巧妙，定時器間隔不是 1s/次，而是跟下次任務的時間相關，這樣就避免了無用的定時器迴圈，也讓執行時間更精準，不存在設定小了浪費資源，設定大了誤差大的情況。接下來進入第二層迴圈。
sort.Sort(byTime(c.entries))
timer = time.NewTimer(c.entries[0].Next.Sub(now))
事件迴圈中，包含了很多事件，比如 新增任務、停止、移除任務，當 cron 啟動之後，這些任務都是非同步的。比如新增任務，不會直接將任務資訊寫入記憶體中，而是進入事件迴圈，加入之後，重新計算第一二層迴圈，避免了正在修改任務資訊，又執行任務資訊，然後出錯的情況。
有人可能會問了，為何不在事件中加鎖，這樣也能避免記憶體競爭。我想說，我們執行的是指令碼任務，有的事件可能很長，可能會阻塞有些事件，所以這些事件都放在迴圈中，避免了加鎖，也滿足了要求。
for {
select {
case now = <-timer.C:
// 執行任務
case newEntry := <-c.add:
// 新增任務
case replyChan := <-c.snapshot:
// 獲取任務資訊
case <-c.stop:
// 停止任務
case id := <-c.remove:
// 移除任務
}
break
}
在瞭解了專案的基本情況之後，對專案做了部分改造，方便使用。
在主迴圈彙總加入了號誌監聽，當觸發號誌 SIGUSR1，將任務資訊輸出到紀錄檔：
usrSig := make(chan os.Signal, 1)
signal.Notify(usrSig, syscall.SIGUSR1)
for {
select {
case <-usrSig:
// 啟動單獨的協程去列印定時任務執行資訊
continue
}
break
}
目前指令碼只能根據指令碼 id 去移除要執行的任務，執行過程中，也不能通過命令去移除任務，不是太方便。比如有個指令碼馬上要執行了，但是該指令碼發現問題了，這時候生產環境的話，就需要更新程式碼，然後重啟服務去下線指令碼任務，這時候，黃花菜可能都涼了。
所以我也是通過號誌，來處理執行之後，執行中移除任務的問題，收到號誌之後，讀取檔案中的內容，根據命令去處理 runing 中的記憶體：
usrSig2 := make(chan os.Signal, 1)
signal.Notify(usrSig2, syscall.SIGUSR2)
......
case <-usrSig2:
actionByte, err := os.ReadFile("/tmp/cron.action")
...... //校驗命令正確性
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])
}
......
由於原專案已經 2 年多沒有個更新過了，就算髮起 pr 估計也不會被處理，所以 fork 一份放在了這裡 aizuyan/cron 進行改造，下面是改進之後的程式碼：
package main
import (
// 載入組態檔
"fmt"
"github.com/aizuyan/cron/v3"
)
func main() {
c := cron.New(cron.WithLogger(cron.DefaultLogger))
c.AddFuncWithTag("流量包過期", "*/5 * * * *", func() {
fmt.Println("你的流量包即將過期了")
})
c.AddFuncWithTag("轉碼包過期", "*/2 * * * *", func() {
fmt.Println("你的轉碼包即將過期了")
})
c.Start()
for {
select {}
}
}
對每個定時任務增加了一個名稱標識，當任務啟動後，當我們執行 kill -SIGUSR1 <pid> 的時候，會看到 stdout 輸出了執行的任務列表資訊：
+----+------------+-------------+---------------------+---------------------+
| ID | TAG | SPEC | PREV | NEXT |
+----+------------+-------------+---------------------+---------------------+
| 2 | 轉碼包過期 | */2 * * * * | 0001-01-01 00:00:00 | 2023-04-02 17:22:00 |
| 1 | 流量包過期 | */5 * * * * | 0001-01-01 00:00:00 | 2023-04-02 17:25:00 |
+----+------------+-------------+---------------------+---------------------+
執行 kill -SIGUSR2 <pid>，移除轉碼包過期任務，避免了使用 ID 容易出錯的問題。
cat /tmp/cron.action
removeTag 轉碼包過期
// {"data":["tag","轉碼包過期"],"level":"info","msg":"removedByTag","time":"2023-04-02T18:32:56+08:00"}
放目前為止，是不是更好用了，基本能滿足我們的需求了，也可以自己去再做各種擴充套件。