Advanced Golang Scheduler (AGScheduler) is a task scheduling library for Golang that supports multiple scheduling types, dynamically changing and persistent jobs, remote call, and cluster
English | 简体中文
- Supports three scheduling types
- One-off execution
- Interval execution
- Cron-style scheduling
- Supports multiple job store methods
- Supports remote call
- gRPC
- HTTP APIs
- Supports cluster
- Remote worker nodes
package main
import (
"context"
"fmt"
"log/slog"
"time"
"github.com/kwkwc/agscheduler"
"github.com/kwkwc/agscheduler/stores"
)
func printMsg(ctx context.Context, 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 {}
}
Since golang can't serialize functions, you need to register them with
RegisterFuncs
beforescheduler.Start()
// Server
srservice := services.SchedulerRPCService{
Scheduler: scheduler,
Address: "127.0.0.1:36360",
}
srservice.Start()
// Client
conn, _ := grpc.Dial("127.0.0.1:36360", grpc.WithTransportCredentials(insecure.NewCredentials()))
client := pb.NewSchedulerClient(conn)
client.AddJob(ctx, job)
// Server
shservice := services.SchedulerHTTPService{
Scheduler: scheduler,
Address: "127.0.0.1:36370",
}
shservice.Start()
// Client
mJob := map[string]any{...}
bJob, _ := json.Marshal(bJob)
resp, _ := http.Post("http://127.0.0.1:36370/scheduler/job", "application/json", bytes.NewReader(bJob))
// Main Node
cnMain := &agscheduler.ClusterNodeClusterNode{
Endpoint: "127.0.0.1:36380",
EndpointHTTP: "127.0.0.1:36390",
SchedulerEndpoint: "127.0.0.1:36360",
Queue: "default",
}
schedulerMain.SetClusterNode(ctx, cnMain)
cserviceMain := &services.ClusterService{Cn: cnMain}
cserviceMain.Start()
// Node
cn := &agscheduler.ClusterNode{
MainEndpoint: "127.0.0.1:36380",
Endpoint: "127.0.0.1:36381",
EndpointHTTP: "127.0.0.1:36391",
SchedulerEndpoint: "127.0.0.1:36361",
Queue: "node",
}
scheduler.SetClusterNode(ctx, cn)
cservice := &services.ClusterService{Cn: cn}
cservice.Start()
gRPC Function | HTTP Method | HTTP Endpoint |
---|---|---|
AddJob | POST | /scheduler/job |
GetJob | GET | /scheduler/job/:id |
GetAllJobs | GET | /scheduler/jobs |
UpdateJob | PUT | /scheduler/job |
DeleteJob | DELETE | /scheduler/job/:id |
DeleteAllJobs | DELETE | /scheduler/jobs |
PauseJob | POST | /scheduler/job/:id/pause |
ResumeJob | POST | /scheduler/job/:id/resume |
RunJob | POST | /scheduler/job/run |
Start | POST | /scheduler/start |
Stop | POST | /scheduler/stop |
RPC Function | HTTP Method | HTTP Endpoint |
---|---|---|
Nodes | GET | /cluster/nodes |