引子

任务调度（Task Scheduling）是很多软件系统中的重要组成部分，字面上的意思是按照一定要求分配运行一些通常时间较长的脚本或程序。在爬虫管理平台 Crawlab[1] 中，任务调度是其中的核心模块，相信不少朋友会好奇如何编写一个任务调度系统。本篇文章会教读者用 Go 语言编写一个非常简单的任务调度系统。

思路

我们首先理清一下思路，开发最小化任务调度器需要什么。

  • 交互界面（API）
  • 定时任务（Cron）
  • 任务执行（Execute Tasks）

整个流程如下：
实战 Go：怎样快速实现一个极简任务调度系统 - 图1
image-20221003094216157
我们通过 API 创建定时任务，执行器根据定时任务标准定期执行脚本。

实战

交互界面

首先我们来搭个架子。在项目目录下创建一个 main.go 文件，并输入以下内容。其中 gin 是非常流行的 Go 语言 API 引擎。

  1. package main
  3. import (
  4.     "fmt"
  5.     "github.com/gin-gonic/gin"
  6.     "os"
  7. )
  9. func main() {
  10.     // api engine
  11.     app := gin.New()
  13.     // api routes
  14.     app.GET("/jobs", GetJobs)
  15.     app.POST("/jobs", AddJob)
  16.     app.DELETE("/jobs", DeleteJob)
  18.     // run api on port 9092
  19.     if err := app.Run(":9092"); err != nil {
  20.       _, err = fmt.Fprintln(os.Stderr, err)
  21.       os.Exit(1)
  22.   }
  23. }

然后添加 api.go 文件，输入以下内容，注意，这里没有任何代码实现，只是加入了占位区域。

  1. package main
  3. import "github.com/gin-gonic/gin"
  5. func GetJobs(c *gin.Context) {
  6.     // TODO: implementation here
  7. }
  9. func AddJob(c *gin.Context) {
  10.     // TODO: implementation here
  11. }
  13. func DeleteJob(c *gin.Context) {
  14.     // TODO: implementation here
  15. }

定时任务

然后是任务调度的核心，定时任务。这里我们使用 robfig/cron[2]，Go 语言比较流行的定时任务库。
现在创建 cron.go 文件，输入以下内容。其中 Cron 就是 robfig/cron 库中的 Cron 类生成的实例。

  1. package main
  3. import "github.com/robfig/cron"
  5. func init() {
  6.     // start to run
  7.     Cron.Run()
  8. }
  10. // Cron create a new cron.Cron instance
  11. var Cron = cron.New()

现在创建好了主要定时任务实例，就可以将核心逻辑添加在刚才的 API 占位区域了。
同样是 api.go ，将核心代码添加进来。

  1. package main
  3. import (
  4.     "github.com/gin-gonic/gin"
  5.     "github.com/robfig/cron/v3"
  6.     "net/http"
  7.     "strconv"
  8. )
  10. func GetJobs(c *gin.Context) {
  11.     // return a list of cron job entries
  12.     var results []map[string]interface{}
  13.     for _, e := range Cron.Entries() {
  14.      results = append(results, map[string]interface{}{
  15.       "id":   e.ID,
  16.       "next": e.Next,
  17.   })
  18.  }
  19.     c.JSON(http.StatusOK, Cron.Entries())
  20. }
  22. func AddJob(c *gin.Context) {
  23.     // post JSON payload
  24.     var payload struct {
  25.      Cron string `json:"cron"`
  26.      Exec string `json:"exec"`
  27.  }
  28.     if err := c.ShouldBindJSON(&payload); err != nil {
  29.      c.AbortWithStatus(http.StatusBadRequest)
  30.      return
  31.  }
  33.     // add cron job
  34.     eid, err := Cron.AddFunc(payload.Cron, func() {
  35.      // TODO: implementation here
  36.  })
  37.     if err != nil {
  38.      c.AbortWithStatusJSON(http.StatusInternalServerError, map[string]interface{}{
  39.       "error": err.Error(),
  40.   })
  41.      return
  42.  }
  44.     c.AbortWithStatusJSON(http.StatusOK, map[string]interface{}{
  45.      "id": eid,
  46.  })
  47. }
  49. func DeleteJob(c *gin.Context) {
  50.     // cron job entry id
  51.     id := c.Param("id")
  52.     eid, err := strconv.Atoi(id)
  53.     if err != nil {
  54.      c.AbortWithStatus(http.StatusBadRequest)
  55.      return
  56.  }
  58.     // remove cron job
  59.     Cron.Remove(cron.EntryID(eid))
  61.     c.AbortWithStatus(http.StatusOK)
  62. }

在这段代码中，我们实现了大部分逻辑，只在 AddJob 的 Cron.AddFunc 中第二个参数里，剩下最后一部分执行任务的代码。下面将来实现一下。

任务执行

现在需要添加任务执行的代码逻辑，咱们创建 exec.go 文件，输入以下内容。这里我们用到了 Go 语言内置的 shell 运行管理库 os/exec，可以执行任何 shell 命令。

  1. package main
  3. import (
  4.     "fmt"
  5.     "os"
  6.     "os/exec"
  7.     "strings"
  8. )
  10. func ExecuteTask(execCmd string) {
  11.     // execute command string parts, delimited by space
  12.     execParts := strings.Split(execCmd, " ")
  14.     // executable name
  15.     execName := execParts[0]
  17.     // execute command parameters
  18.     execParams := strings.Join(execParts[1:], " ")
  20.     // execute command instance
  21.     cmd := exec.Command(execName, execParams)
  23.     // run execute command instance
  24.     if err := cmd.Run(); err != nil {
  25.      _, err = fmt.Fprintln(os.Stderr, err)
  26.      fmt.Println(err.Error())
  27.  }
  28. }

好了，现在我们将这部分执行代码逻辑放到之前的占位区域中。

  1. ...
  2. // add cron job
  3. eid, _ := Cron.AddFunc(payload.Cron, func() {
  4.      ExecuteTask(payload.Exec)
  5.  })
  6. ...

代码效果

OK，大功告成！现在我们可以试试运行这个极简的任务调度器了。
在命令行中敲入 go run .，API 引擎就启动起来了。

  1. [GIN-debug] [WARNING] Running in "debug" mode. Switch to "release" mode in production.
  2.  - using env:   export GIN_MODE=release
  3.  - using code:  gin.SetMode(gin.ReleaseMode)
  5. [GIN-debug] GET    /jobs                     --> main.GetJobs (1 handlers)
  6. [GIN-debug] POST   /jobs                     --> main.AddJob (1 handlers)
  7. [GIN-debug] DELETE /jobs/:id                 --> main.DeleteJob (1 handlers)
  8. [GIN-debug] [WARNING] You trusted all proxies, this is NOT safe. We recommend you to set a value.
  9. Please check https://pkg.go.dev/github.com/gin-gonic/gin#readme-don-t-trust-all-proxies for details.
  10. [GIN-debug] Listening and serving HTTP on :9092

现在打开另一个命令行窗口，输入 curl -X POST -d ‘{“cron”:” *”,”exec”:”touch /tmp/hello.txt”}’ http://localhost:9092/jobs，会得到如下返回结果。表示已经生成了相应的定时任务，任务 ID 为 1，每分钟跑一次，会更新一次 /tmp/hello.txt。

  1. {"id":1}

在这个命令行窗口中输入 curl http://localhost:9092/jobs。

  1. [{"id":1,"next":"2022-10-03T12:46:00+08:00"}]

这表示下一次执行是 1 分钟之后。
等待一分钟，执行 ls -l /tmp/hello.txt，得到如下结果。

  1. -rw-r--r--  1 marvzhang  wheel     0B Oct  3 12:46 /tmp/hello.txt

也就是说，执行成功了，大功告成！

总结

本篇文章通过将 Go 语言几个库简单组合，就开发出了一个极简的任务调度系统。所用到的核心库:

  • gin[3]
  • robfig/cron[4]
  • os/exec

整个代码示例仓库在 GitHub 上: https://github.com/tikazyq/codao-code/tree/main/2022-10/go-task-scheduler