以前一直使用 Python 的任务调度库 APScheduler（支持任务持久化，支持多种存储方式），但由于没有找到与它功能和使用方式类似的 Golang 库，所以模仿 APScheduler 3.x 写了个简易版本的 AGScheduler。

AGScheduler

Advanced Golang Scheduler (AGScheduler) 是一款适用于 Golang 的任务调度程序，支持多种调度方式，支持动态更改和持久化任务，支持远程调用

English | 简体中文

特性

• 支持三种调度方式
  • 一次性执行
  • 间隔执行
  • Cron 式调度
• 支持多种任务存储方式
  • Memory
  • GROM(任何 GROM 支持的 RDBMS 都能运行)
  • Redis
  • MongoDB
  • etcd
• 支持远程调用
  • gRPC
  • HTTP APIs

使用

package main

import (
    "fmt"
    "log/slog"
    "time"

    "github.com/kwkwc/agscheduler"
    "github.com/kwkwc/agscheduler/stores"
)

func printMsg(j agscheduler.Job) {
    slog.Info(fmt.Sprintf("Run job `%s` %s\n\n", j.FullName(), j.Args))
}

func main() {
    agscheduler.RegisterFuncs(printMsg)

    store := &stores.MemoryStore{}
    scheduler := &agscheduler.Scheduler{}
    scheduler.SetStore(store)

    job1 := agscheduler.Job{
        Name:     "Job1",
        Type:     agscheduler.TYPE_INTERVAL,
        Interval: "2s",
        Timezone: "UTC",
        Func:     printMsg,
        Args:     map[string]any{"arg1": "1", "arg2": "2", "arg3": "3"},
    }
    job1, _ = scheduler.AddJob(job1)
    slog.Info(fmt.Sprintf("%s.\n\n", job1))

    job2 := agscheduler.Job{
        Name:     "Job2",
        Type:     agscheduler.TYPE_CRON,
        CronExpr: "*/1 * * * *",
        Timezone: "Asia/Shanghai",
        FuncName: "main.printMsg",
        Args:     map[string]any{"arg4": "4", "arg5": "5", "arg6": "6", "arg7": "7"},
    }
    job2, _ = s.AddJob(job2)
    slog.Info(fmt.Sprintf("%s.\n\n", job2))

    job3 := agscheduler.Job{
        Name:     "Job3",
        Type:     agscheduler.TYPE_DATETIME,
        StartAt:  "2023-09-22 07:30:08",
        Timezone: "America/New_York",
        Func:     printMsg,
        Args:     map[string]any{"arg8": "8", "arg9": "9"},
    }
    job3, _ = s.AddJob(job3)
    slog.Info(fmt.Sprintf("%s.\n\n", job3))

    jobs, _ := s.GetAllJobs()
    slog.Info(fmt.Sprintf("Scheduler get all jobs %s.\n\n", jobs))

    scheduler.Start()

    select {}
}

注册函数

由于 golang 无法序列化函数，所以 scheduler.Start() 之前需要使用 RegisterFuncs 注册函数

gRPC

// Server
rservice := services.SchedulerRPCService{Scheduler: scheduler}
rservice.Start("127.0.0.1:36363")

// Client
conn, _ := grpc.Dial("127.0.0.1:36363", grpc.WithTransportCredentials(insecure.NewCredentials()))
client := pb.NewSchedulerClient(conn)
client.AddJob(ctx, job)

HTTP APIs

// Server
hservice := services.SchedulerHTTPService{Scheduler: scheduler}
hservice.Start("127.0.0.1:63636")

// Client
mJob := map[string]any{...}
bJob, _ := json.Marshal(bJob)
resp, _ := http.Post("http://127.0.0.1:63636/scheduler/job", "application/json", bytes.NewReader(bJob))

示例

完整示例

致谢

APScheduler