job

package
v1.9.0 Latest Latest
Warning

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

Go to latest
Published: Aug 9, 2026 License: MIT Imports: 13 Imported by: 0

Documentation

Overview

Package job 提供由 gocron 驱动的定时任务调度器,实现 go-zero 的 service.Service,可直接加入 service group 与 RPC/API 服务合并部署。

服务侧只需要提供配置和一组具名 handler,并发控制、单例、超时、panic 恢复、优雅退出都由本包统一处理。

Index

Constants

This section is empty.

Variables

View Source
var (
	// ErrScheduleRequired 表示 job 既没有配置 cron 也没有配置 every。
	ErrScheduleRequired = errors.New("job: cron or every is required")

	// ErrScheduleConflict 表示 job 同时配置了 cron 和 every。
	ErrScheduleConflict = errors.New("job: cron and every are mutually exclusive")

	// ErrInvalidEvery 表示 every 不是正数。
	ErrInvalidEvery = errors.New("job: every must be positive")

	// ErrInvalidTimeout 表示 timeout 为负数。
	ErrInvalidTimeout = errors.New("job: timeout must not be negative")

	// ErrInvalidOverlap 表示 overlap 取值不在 allow/skip/wait 之内。
	ErrInvalidOverlap = errors.New("job: overlap must be one of allow, skip, wait")

	// ErrInvalidLimitMode 表示并发 mode 取值不在 skip/wait 之内。
	ErrInvalidLimitMode = errors.New("job: concurrency mode must be one of skip, wait")

	// ErrInvalidConcurrencyLimit 表示并发上限为负数。
	ErrInvalidConcurrencyLimit = errors.New("job: concurrency limit must not be negative")

	// ErrOverlapWaitWithConcurrency 表示全局并发限制与 job wait 队列组合使用。
	// gocron 的全局 limiter 优先于 singleton limiter,这个组合会静默丢失 tick。
	ErrOverlapWaitWithConcurrency = errors.New("job: overlap wait cannot be combined with a global concurrency limit")

	// ErrInvalidCron 表示 cron 表达式无法解析或没有未来触发时间。
	ErrInvalidCron = errors.New("job: invalid cron expression")

	// ErrEmptyHandlerName 表示注册了空名字的 handler。
	ErrEmptyHandlerName = errors.New("job: handler name must not be empty")

	// ErrNilHandler 表示注册了空 handler。
	ErrNilHandler = errors.New("job: handler must not be nil")

	// ErrDuplicateHandler 表示同一个名字注册了多次。
	ErrDuplicateHandler = errors.New("job: duplicate handler")

	// ErrHandlerNotFound 表示配置里声明的 job 没有对应的 handler。
	ErrHandlerNotFound = errors.New("job: configured job has no handler")

	// ErrJobNotConfigured 表示注册的 handler 在配置里找不到对应的 job。
	ErrJobNotConfigured = errors.New("job: registered handler is not configured")

	// ErrPanic 表示 handler 内部 panic 已被恢复。
	ErrPanic = errors.New("job: panic recovered")
)

Functions

func WithNamespace

func WithNamespace(namespace string) opts.Opt[Options]

WithNamespace 设置 job 名前缀,最终 job 名为 "<namespace>.<name>"。

Types

type ConcurrencyConf

type ConcurrencyConf struct {
	// Limit 为 0 表示不限制,防重叠请优先用 Spec.Overlap。
	Limit int `json:",optional"`

	// Mode 只在 Limit > 0 时有意义,留空取 wait。
	Mode LimitMode `json:",optional"`
}

ConcurrencyConf 是调度器级别的并发限制。

type Config

type Config struct {
	// Enable 决定 JobServer 是否加入 service group,由调用方判断。
	Enable bool `json:",optional"`

	// Timezone 为 cron 表达式的时区,留空表示 time.Local。
	Timezone string `json:",optional"`

	// ShutdownTimeout 是等待运行中 job 结束的时间,留空取 3s。
	ShutdownTimeout time.Duration `json:",optional"`

	// Concurrency 是调度器级别的全局并发限制。
	Concurrency ConcurrencyConf `json:",optional"`

	// Jobs 以 job 名为 key,必须与注册的 handler 严格一一对应。
	Jobs map[string]Spec `json:",optional"`
}

Config 是 job 调度器的配置,直接对应服务 yaml 中的 job 段。

type Handler

type Handler func(context.Context) error

Handler 是一次 job 执行。ctx 会在进程退出时被取消, 配置了 timeout 时还会带上超时,handler 应当响应取消。

type LimitMode

type LimitMode string

LimitMode 描述调度器整体并发达到上限时的行为。

const (
	// LimitModeSkip 丢弃超出并发上限的这次触发。
	LimitModeSkip LimitMode = "skip"

	// LimitModeWait 排队等待空闲槽位。
	LimitModeWait LimitMode = "wait"
)

type NamedHandler

type NamedHandler struct {
	Name    string
	Handler Handler
}

NamedHandler 把 job 名与 handler 绑定,名字必须与配置里的 key 一致。

func Named

func Named(name string, handler Handler) NamedHandler

Named 构造一个 NamedHandler。

type Options

type Options struct {
	// Namespace 会作为 job 名的前缀,通常传服务名。
	// 分布式锁以 job 名作为 key,没有前缀时不同服务里的同名 job 会互相抢锁。
	Namespace string
}

func (Options) DefaultOptions

func (o Options) DefaultOptions() Options

type OverlapPolicy

type OverlapPolicy string

OverlapPolicy 描述同一个 job 上一次还没跑完时的行为。

const (
	// OverlapAllow 允许同一个 job 并发执行。
	OverlapAllow OverlapPolicy = "allow"

	// OverlapSkip 丢弃本次触发,等下一个周期。
	OverlapSkip OverlapPolicy = "skip"

	// OverlapWait 排队等上一次执行结束。
	OverlapWait OverlapPolicy = "wait"
)

type Server

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

Server 是 job 调度器,实现 service.Service。

func New

func New(cfg Config, handlers []NamedHandler, op ...opts.Opt[Options]) (*Server, error)

New 校验配置与 handler 并构建调度器。配置里的 job 与注册的 handler 必须严格一一对应,任何一边多出或缺失都会返回错误,避免改了 yaml 忘了写代码(或反过来)时任务静默不执行。

func (*Server) Start

func (s *Server) Start()

Start 实现 service.Service,非阻塞。

func (*Server) Stop

func (s *Server) Stop()

Stop 实现 service.Service,取消所有运行中 job 的 ctx 并等待其结束。

type Spec

type Spec struct {
	Enable bool `json:",default=true"`

	// Cron 为 5 位或 6 位(含秒)表达式,也支持 @every 5s 这类描述符。
	Cron string `json:",optional"`

	// Every 表示固定间隔调度。
	Every time.Duration `json:",optional"`

	// Overlap 为上次未结束时的行为,留空取 skip。
	Overlap OverlapPolicy `json:",optional"`

	// Timeout 为单次执行的超时,留空表示不限制。超时只会取消 ctx,
	// handler 必须自己响应取消才能真正停下来。
	Timeout time.Duration `json:",optional"`
}

Spec 是单个 job 的配置。Cron 与 Every 二选一。

Jump to

Keyboard shortcuts

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