cron — 定时任务与选主

多副本部署时，一个定时任务应该在每个周期只跑一次，而不是每个副本各跑一次。 pkg/core/cron 的做法是：每个副本都起自己的定时器，到点时去抢一个以 job 命名的 redis 认领（claim），抢到的那个执行。这个示例注册了四个不同形态的 job， 并暴露一个计数接口，让「到底跑了几次」从外部看得见。

运行

go run ./examples/cron

curl -s localhost:8084/api/v1/jobs     # 每个 job 跑了多少次、最后什么时候跑的
curl -s localhost:8084/version         # 内置的构建信息端点

监听 :8084，-conf 默认指向 examples/cron/config.yaml。默认配置里 redis 段是注释掉的， 因此单独跑不需要 make dev——不配 redis 就是单副本模式，每个 job 直接跑。 要看选主效果需要 make dev 起 redis 并打开 config.yaml 里的 redis: 段。

它演示了什么

• 三种调度形态：Interval 固定周期、Schedule cron 表达式、At/Delay 一次性任务。
• RunOnStart：Interval 专属，让 job 在启动时立刻跑一次，而不是等满第一个周期。
• 失败不会停循环：reconcile 每第三次返回 error，下一轮照常触发。
• 单次运行有上限：cron.Config.Timeout（示例配置是 2s）界定一次 run 的最长时长。
• 认领持有整个时间槽：Interval 同时也是 claim 的 TTL，所以各副本定时器不同步也不会重复执行。
• 挂到 env.Block 上：verticle 会在关闭基础设施之前先停掉 job。

注册四个 job

这段来自 examples/cron/main.go 的 jobs 方法。它是一个 收参数的方法而不是写死的字面量，就是为了让测试能在更短的时钟上构建同一组 job。 四个 job 正好对应三种调度形态。

func (s *cronService) jobs(heartbeat, reconcile time.Duration, report string, notify time.Duration) []cron.Job {
	return []cron.Job{
		{
			Name:     "heartbeat",
			Interval: heartbeat,
			// Run once at startup so the first counter bump is immediate.
			RunOnStart: true,
			Run: func(ctx context.Context) error {
				s.stats.record("heartbeat")
				obs.Info(ctx, "heartbeat")
				return nil
			},
		},
		{
			Name:     "reconcile",
			Interval: reconcile,
			Run: func(context.Context) error {
				// Every third run fails, to show that one bad run does not
				// stop the loop — the next tick still happens.
				if n := s.stats.record("reconcile"); n%3 == 0 {
					return errors.New("reconcile: the batch was rejected")
				}
				return nil
			},
		},
		{
			Name:     "report",
			Schedule: report,
			Run: func(ctx context.Context) error {
				s.stats.record("report")
				obs.Info(ctx, "report generated")
				return nil
			},
		},
		{
			Name:  "notify",
			Delay: notify,
			Run: func(ctx context.Context) error {
				s.stats.record("notify")
				obs.Info(ctx, "one-shot notify fired")
				return nil
			},
		},
	}
}

装配与启动

同样来自 examples/cron/main.go。注意 env.Redis() 是 逗号-ok 形式：没配 redis 时 rc 为 nil、ok 为 false， 直接传给 cron.Config.Redis 就表示单副本模式，调用点不需要判空。

	s.stats = newJobStats()
	rc, ok := env.Redis()
	s.redis = rc
	s.sched = cron.New(cron.Config{
		// nil when redis is not configured: one replica, no election needed.
		Redis:   rc,
		Timeout: s.cfg.JobTimeout,
	})
	s.sched.Add(s.jobs(s.cfg.HeartbeatInterval, s.cfg.ReconcileInterval, s.cfg.DailyReport, s.cfg.NotifyDelay)...)
	obs.Info(ctx, "cron jobs registered",
		"jobs", len(s.sched.Jobs()), "leader_election", ok)

	// verticle starts this block and cancels it on shutdown, before the
	// infrastructure it uses is closed.
	env.Block("cron", s.sched.Run)

为什么认领要活满一整个槽

sequenceDiagram
    participant A as 副本 A
(t=0,10,20…)
    participant R as redis claim
    participant B as 副本 B
(t=5,15,25…)
    A->>R: t=0 抢 claim（TTL = interval）
    R-->>A: 抢到 → 执行
    B->>R: t=5 抢 claim
    R-->>B: 已被占用 → 跳过（不执行）
    Note over R: t=10 claim 过期
    A->>R: t=10 重新抢到 → 执行
    B->>R: t=15 抢 claim
    R-->>B: 已被占用 → 跳过
				

推论有三条：Job.Interval 同时是 claim 的 TTL，run 必须远小于 interval； cron 表达式的 claim TTL 是「这个槽的长度」（下次触发减本次触发）； 一次性任务相反——它的 claim 永不过期，所以跨副本最多执行一次， 代价是失败也不会自动重试。

验证

跑大约 10 秒后看计数：

curl -s localhost:8084/api/v1/jobs

{
  "code": 0,
  "msg": "ok",
  "data": {
    "jobs": {
      "heartbeat": { "runs": 48, "last": "2026-…T…Z" },
      "reconcile": { "runs": 10, "last": "2026-…T…Z" },
      "report":    { "runs": 2,  "last": "2026-…T…Z" },
      "notify":    { "runs": 1,  "last": "2026-…T…Z" }
    }
  }
}

具体数字随运行时长变化，但形状是固定的：heartbeat 按 heartbeat_interval（示例配成 200ms）增长、 reconcile 按 reconcile_interval（1s）增长、 report 每 5 秒加一（配置里 daily_report: "@every 5s"）， 而 notify 只会出现一次。看选主效果：

go run ./examples/cron &       # 副本 A
go run ./examples/cron &       # 副本 B（同一个 job 名 → 同一把 claim）
sleep 2
curl -s localhost:8084/api/v1/jobs   # 计数按 interval 增长，而不是两倍

测试

测试不需要 Docker：redistest.Start(t) 起进程内的 miniredis， claim 就能工作。它钉住的具体事实：

• TestJobsRunOnTheirIntervals：把 interval 压到 5ms 后， heartbeat 与 reconcile 的计数都至少到 2； 取消 ctx 后调度器必须停下。
• TestRunOnStartBumpsImmediately：interval 设成一小时时， heartbeat 立刻是 1，而没有 RunOnStart 的 reconcile 仍然是 0。
• TestJobFailureDoesNotStopTheLoop：手动跑 6 次 reconcile， 恰好失败 2 次（每第三次），计数仍是 6——失败不掉 job。
• TestLeaderElectionRunsOneInstancePerInterval：两个副本共享 miniredis， A 先抢到 claim，B 启动后抢不到；两个副本的 heartbeat 计数之和恰好是 1。
• TestOneShotLeaderElection：错峰启动的两个副本跑同一个一次性任务， 合计也恰好是 1。
• TestConfigDefaults：默认值是 500ms / 2s / "@every 5s" / 10s / 5s， 注册的 job 名字依次是 heartbeat, reconcile, report, notify， 且无 redis 段时 svc.redis 为 nil。
• TestCronJobFiresOnItsSchedule：表达式改成 "@every 1s" （robfig 真实时钟能支持的最快档）后，report 在 3 秒内至少跑 2 次。
• TestOneShotFiresExactlyOnce：一次性任务跑完一次后保持沉默。
• TestJobsEndpointReportsCounters：接口返回的 data.jobs 里计数与 last 都对得上。

go test ./examples/cron

相关

相关示例

• examples/lifecycle：Block 与关闭顺序。
• examples/observability：job 的日志往哪走。
• examples/resilience：单次运行失败时的重试与熔断。