Help us shape the future of Go! Take the 2026 Go Developer Survey

local

package
v0.2.5 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 29, 2026 License: MIT Imports: 8 Imported by: 0

Documentation

Overview

Package local 提供纯本地的 cron 调度器:按 cron 表达式触发任务。

零外部中间件、零 Redis 依赖——依赖面只有标准库与 github.com/oekit/pkg/timex/cron(cron 表达式解析复用 oekit 矩阵的实现, 不重复造轮子)。适合纯 Go 嵌入式服务形态(CGO_ENABLED=0 静态链接、 SQLite 落盘、零外部中间件)——这类形态用不了 Redis 后端, 但仍需要「到点自动干活」的能力。

与根包的关系

本包是 github.com/oekit/timertask 的子包(同模块),但不 import 根包: 两者互不依赖(根包也不依赖本包),各自独立演进。需要「延迟队列」 (投递到未来某时刻)而不是「cron 触发」时,用根包的 LocalBackend。

语义(明确约定)

并发度:

  • Config.Concurrency 是全局并发度(同时执行的任务数上限),默认 1(严格串行)。
  • 同一 spec 的触发始终串行(自身永不重叠):上一轮没跑完时, 新的触发进入该 spec 的待执行队列。

错过触发点(进程停机、时钟跳变、上一轮未跑完):

  • Config.CatchUp = CatchUpSkip(默认):只执行最近的一个触发点, 其余错过点全部跳过;待执行队列里的旧触发会被新触发替换 (同一 spec 最多积压 1 个)——不会因为任务慢而无限堆积。
  • Config.CatchUp = CatchUpAll:所有错过的触发点按时间顺序逐一补跑。

进程退出:

  • Stop 不打断正在执行的任务:任务的 ctx 不会被取消,Stop 等待它们结束; Stop 的等待上限由调用方给的 ctx 决定,超时返回 ctx.Err(), 任务继续跑完(Go 无法安全中断 goroutine)。
  • Start 的 ctx 被取消时:调度立刻停止,任务的 ctx 同时被取消 (任务可自行感知并尽快退出),未开始的触发被丢弃。
  • 两种情况下「未开始的触发」都会被丢弃——本地调度器不持久化。

重复注册:

  • 同一个 spec 重复 Schedule 返回 ErrDuplicateSpec(不静默覆盖: 覆盖会悄悄丢掉原处理函数)。要换处理函数请先 Unschedule。

示例

sched, err := local.New(local.Config{Concurrency: 2})
if err != nil {
	return err
}
if err := sched.Schedule("0 9 * * 1-5", func(ctx context.Context, at time.Time) error {
	return report(ctx)
}); err != nil {
	return err
}
go func() { _ = sched.Start(ctx) }()
defer sched.Stop(context.Background())

Index

Constants

This section is empty.

Variables

View Source
var ErrDuplicateSpec = errors.New("local: cron spec already scheduled")

ErrDuplicateSpec 表示同一个 cron 表达式被重复注册。

Functions

This section is empty.

Types

type CatchUpPolicy

type CatchUpPolicy int

CatchUpPolicy 是错过触发点的补偿策略。

const (
	// CatchUpSkip(默认):只执行最近的一个触发点。
	//
	// 适用于「状态收敛型」任务(对账、刷新、巡检)——历史触发点已过时,
	// 补跑没有意义,甚至有害(重复处理旧状态)。
	CatchUpSkip CatchUpPolicy = iota
	// CatchUpAll:所有错过的触发点按时间顺序逐一补跑。
	//
	// 适用于「计数型」任务(每次触发都代表一次必须发生的动作)。
	// 注意补跑会积压:任务执行慢于触发周期时,积压会持续增长。
	CatchUpAll
)

type Clock

type Clock interface {
	// Now 返回当前时间。
	Now() time.Time
	// After 返回一个通道:在 d 之后收到一个时间信号。
	// d <= 0 时必须返回「已就绪」的通道(调用方据此立即重算,不空等)。
	After(d time.Duration) <-chan time.Time
}

Clock 是时钟抽象,便于注入假时钟做确定性测试。

调度器的所有时间判断都走本接口:注册时的基准时间、触发点计算、等待下一个 触发点。注入 FakeClock 后,整个调度过程不需要 sleep 即可断言触发序列。

type Config

type Config struct {
	// Concurrency 全局任务并发度(同时执行的任务数上限),<=0 取 1(严格串行)。
	//
	// 同一 spec 的触发始终串行(自身不重叠);不同 spec 之间按此上限并发。
	Concurrency int
	// CatchUp 错过触发点的补偿策略,零值取 CatchUpSkip。
	CatchUp CatchUpPolicy
	// Clock 时钟注入,nil 取 SystemClock{};测试可注入 FakeClock。
	Clock Clock
	// Logger 日志器,nil 取 slog.Default()。
	Logger *slog.Logger
	// Location cron 表达式的时区,nil 取 time.Local。
	Location *time.Location
	// MaxCatchUp 单次补跑的最大触发点数,<=0 取 1000。
	MaxCatchUp int
}

Config 是 Scheduler 配置。

type FakeClock

type FakeClock struct {
	// contains filtered or unexported fields
}

FakeClock 是可手动推进的假时钟,用于确定性测试(不依赖 sleep)。

典型用法:

clock := local.NewFakeClock(time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC))
sched, _ := local.New(local.Config{Clock: clock, Location: time.UTC})
sched.Schedule("*/5 * * * * *", fn)
go sched.Start(ctx)
clock.Advance(5 * time.Second) // 触发点到达,fn 被调用

func NewFakeClock

func NewFakeClock(t time.Time) *FakeClock

NewFakeClock 创建以 t 为当前时间的假时钟。

func (*FakeClock) Advance

func (c *FakeClock) Advance(d time.Duration)

Advance 推进假时钟 d,并唤醒所有已到期的等待者。

func (*FakeClock) After

func (c *FakeClock) After(d time.Duration) <-chan time.Time

After 返回等待通道:假时钟推进到该等待的截止时间时,通道收到信号。

d <= 0 时立即返回已就绪的通道(不注册等待者)。

func (*FakeClock) Now

func (c *FakeClock) Now() time.Time

Now 返回假时钟的当前时间。

type Scheduler

type Scheduler struct {
	// contains filtered or unexported fields
}

Scheduler 是基于 cron 表达式的本地调度器。

零外部中间件、零 Redis 依赖:只依赖标准库与 github.com/oekit/pkg/timex/cron (cron 表达式解析复用 oekit 矩阵的实现,不重复造轮子)。

典型用法:

sched, err := local.New(local.Config{Concurrency: 4})
if err != nil {
	return err
}
if err := sched.Schedule("*/5 * * * * *", func(ctx context.Context, at time.Time) error {
	return doSomething(ctx)
}); err != nil {
	return err
}
go func() { _ = sched.Start(ctx) }()
defer sched.Stop(context.Background())

func New

func New(cfg Config) (*Scheduler, error)

New 创建调度器。

func (*Scheduler) Schedule

func (s *Scheduler) Schedule(spec string, fn Task) error

Schedule 注册一个 cron 任务。

spec 交给 github.com/oekit/pkg/timex/cron 解析:支持 5 字段(分 时 日 月 周) 与 6 字段(秒 分 时 日 月 周)格式,以及 @every/@daily/@hourly 等宏。

同一个 spec 重复注册返回 ErrDuplicateSpec(不静默覆盖:覆盖会悄悄丢掉 原处理函数,且调用方难以察觉)。要换处理函数请先 Unschedule。

func (*Scheduler) Start

func (s *Scheduler) Start(ctx context.Context) error

Start 启动调度循环并阻塞,直到 ctx 取消或 Stop 被调用。

返回时调度器已停止:不再产生新触发;正在执行的任务不会被强制中断 (任务可通过 ctx 感知取消并自行退出)。重复调用返回错误。

func (*Scheduler) Stop

func (s *Scheduler) Stop(ctx context.Context) error

Stop 停止调度并等待正在执行的任务结束。

语义:

  • 未开始的触发被丢弃;
  • 正在执行的任务不被打断(任务的 ctx 不被取消),Stop 最多等待到 ctx 截止;
  • ctx 超时返回 ctx.Err(),任务继续跑完(Go 无法安全中断 goroutine)。

可重复调用。

func (*Scheduler) Unschedule

func (s *Scheduler) Unschedule(spec string) bool

Unschedule 注销一个 cron 任务,返回是否确实移除了该任务。

正在执行的任务不会被中断(任务体可自行感知 ctx 取消);未开始的触发被丢弃。

type SystemClock

type SystemClock struct{}

SystemClock 是真实时钟(time.Now + time.After)。

func (SystemClock) After

func (SystemClock) After(d time.Duration) <-chan time.Time

After 返回一个在 d 之后收到信号的通道。

func (SystemClock) Now

func (SystemClock) Now() time.Time

Now 返回当前时间。

type Task

type Task func(ctx context.Context, at time.Time) error

Task 是被 cron 触发的任务函数。

at 为本次触发的计划时间(cron 计算出的到期点),可用于日志与幂等键。 ctx 是调度器的运行上下文(Start 的 ctx),取消后任务应尽快返回。

任务返回错误只记日志、不重试:需要重试语义请在任务内部实现, 或改用 github.com/oekit/timertask 的 LocalBackend 延迟队列。

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL