core

package
v1.7.1 Latest Latest
Warning

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

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

Documentation

Overview

Package jobx 提供单进程任务执行与调度库：异步任务、延迟任务与 定时任务，与 errx / logx 生态打通。

jobx 不是消息队列，不解决跨进程消息传递；它解决单进程内每个业务 都要重复的部分：任务队列、worker 池、延迟执行、失败重试、同名任务 冲突策略与优雅关闭。

Index

Constants

View Source 
const (
	// CodeInvalidConfig 配置非法。
	CodeInvalidConfig errx.Code = "jobx_invalid_config"
	// CodeHandlerNotFound 未注册处理器。
	CodeHandlerNotFound errx.Code = "jobx_handler_not_found"
	// CodeHandlerConflict 处理器重复注册。
	CodeHandlerConflict errx.Code = "jobx_handler_conflict"
	// CodeJobInvalid 任务参数非法。
	CodeJobInvalid errx.Code = "jobx_job_invalid"
	// CodeJobNotFound 任务或调度条目不存在。
	CodeJobNotFound errx.Code = "jobx_job_not_found"
	// CodeQueueFull 就绪队列已满（丢弃策略）。
	CodeQueueFull errx.Code = "jobx_queue_full"
	// CodeShuttingDown 关闭中拒绝新任务。
	CodeShuttingDown errx.Code = "jobx_shutting_down"
	// CodeTimeout 单任务执行超时。
	CodeTimeout errx.Code = "jobx_timeout"
	// CodeRetryExhausted 重试耗尽。
	CodeRetryExhausted errx.Code = "jobx_retry_exhausted"
	// CodeExecutionFailed 处理器执行失败（含 panic）。
	CodeExecutionFailed errx.Code = "jobx_execution_failed"
	// CodeSkipped 同名任务在途，本次提交被跳过。
	CodeSkipped errx.Code = "jobx_skipped"
	// CodeReplaced 同名旧任务已被替换取消。
	CodeReplaced errx.Code = "jobx_replaced"
	// CodeCronInvalid cron 表达式非法。
	CodeCronInvalid errx.Code = "jobx_cron_invalid"
	// CodeSchedulerStopped 调度器已停止。
	CodeSchedulerStopped errx.Code = "jobx_scheduler_stopped"
	// CodeStoreInvalid 任务存储读写失败。
	CodeStoreInvalid errx.Code = "jobx_store_invalid"
	// CodeIDGenerateFailed 任务 ID 生成失败。
	CodeIDGenerateFailed errx.Code = "jobx_id_generate_failed"
	// CodeJobCancelled 任务已取消。
	CodeJobCancelled errx.Code = "jobx_job_cancelled"
)

错误码统一以 jobx_ 为前缀。

View Source 
const Version = "v1.7.1"

Version 是当前库版本，与 git tag 保持一致。

Variables

View Source 
var (
	// ErrInvalidConfig 配置非法。
	ErrInvalidConfig = errx.NewCode(CodeInvalidConfig, "配置非法")
	// ErrHandlerNotFound 处理器未注册。
	ErrHandlerNotFound = errx.NewCode(CodeHandlerNotFound, "处理器未注册")
	// ErrHandlerConflict 处理器重复注册。
	ErrHandlerConflict = errx.NewCode(CodeHandlerConflict, "处理器重复注册")
	// ErrJobInvalid 任务参数非法。
	ErrJobInvalid = errx.NewCode(CodeJobInvalid, "任务参数非法")
	// ErrJobNotFound 任务或调度条目不存在。
	ErrJobNotFound = errx.NewCode(CodeJobNotFound, "任务或调度条目不存在")
	// ErrQueueFull 任务队列已满。
	ErrQueueFull = errx.NewCode(CodeQueueFull, "任务队列已满")
	// ErrShuttingDown 执行器关闭中。
	ErrShuttingDown = errx.NewCode(CodeShuttingDown, "执行器关闭中")
	// ErrTimeout 任务执行超时。
	ErrTimeout = errx.NewCode(CodeTimeout, "任务执行超时")
	// ErrRetryExhausted 任务重试耗尽。
	ErrRetryExhausted = errx.NewCode(CodeRetryExhausted, "任务重试耗尽")
	// ErrExecutionFailed 处理器执行失败（含 panic）。
	ErrExecutionFailed = errx.NewCode(CodeExecutionFailed, "处理器执行失败")
	// ErrSkipped 同名任务在途，本次提交被跳过。
	ErrSkipped = errx.NewCode(CodeSkipped, "同名任务在途，本次提交被跳过")
	// ErrReplaced 同名旧任务已被替换取消。
	ErrReplaced = errx.NewCode(CodeReplaced, "同名旧任务已被替换取消")
	// ErrCronInvalid cron 表达式非法。
	ErrCronInvalid = errx.NewCode(CodeCronInvalid, "cron 表达式非法")
	// ErrSchedulerStopped 调度器已停止。
	ErrSchedulerStopped = errx.NewCode(CodeSchedulerStopped, "调度器已停止")
	// ErrStoreInvalid 任务存储读写失败。
	ErrStoreInvalid = errx.NewCode(CodeStoreInvalid, "任务存储读写失败")
)

预定义错误值，可用 errx.Is / errors.Is 判断。

Functions

This section is empty.

Types

type ConflictPolicy 

type ConflictPolicy uint8

ConflictPolicy 同名任务在途时的处理策略。 

const (
	// ConflictSkip 跳过新任务，返回 ErrSkipped（默认）。
	ConflictSkip ConflictPolicy = iota
	// ConflictReplace 取消同名旧任务并执行新任务（尽力取消）。
	ConflictReplace
	// ConflictAllow 允许同名任务并发执行。
	ConflictAllow
)

type Dispatcher 

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

Dispatcher 任务执行器：就绪队列 + worker 池 + 延迟队列。 所有方法并发安全。

func NewDispatcher 

func NewDispatcher(opts ...Option) (*Dispatcher, error)

NewDispatcher 构造执行器并启动 worker 池与延迟调度。

func (*Dispatcher) Cancel 

func (d *Dispatcher) Cancel(id string) error

Cancel 取消任务：延迟/排队中的直接取消，执行中的通过 context 协作。

func (*Dispatcher) Handle 

func (d *Dispatcher) Handle(name string, h Handler) error

Handle 注册任务处理器；空名/超长名返回 ErrJobInvalid， 重复注册返回 ErrHandlerConflict。

func (*Dispatcher) JobStatus 

func (d *Dispatcher) JobStatus(id string) (Status, error)

JobStatus 查询任务状态；未知 ID 返回 ErrJobNotFound。

func (*Dispatcher) Restore 

func (d *Dispatcher) Restore(ctx context.Context) (int, error)

Restore 从持久化存储恢复未完成任务（进程重启后调用），返回恢复数量。 恢复绕过冲突检查但重建在途集合；缺失处理器的任务被删除并跳过。

func (*Dispatcher) Shutdown 

func (d *Dispatcher) Shutdown(ctx context.Context) error

Shutdown 优雅关闭：拒绝新提交、丢弃延迟任务、等待存量执行完毕； ctx 超时则取消执行中的任务并返回 ctx 错误。幂等。

func (*Dispatcher) Submit 

func (d *Dispatcher) Submit(ctx context.Context, name string, payload []byte, opts ...SubmitOption) (string, error)

Submit 提交任务（立即执行），返回任务 ID。

func (*Dispatcher) SubmitAfter 

func (d *Dispatcher) SubmitAfter(ctx context.Context, name string, payload []byte, delay time.Duration, opts ...SubmitOption) (string, error)

SubmitAfter 提交延时执行的任务（delay 必须非负）。

func (*Dispatcher) SubmitAt 

func (d *Dispatcher) SubmitAt(ctx context.Context, name string, payload []byte, at time.Time, opts ...SubmitOption) (string, error)

SubmitAt 提交指定时刻执行的任务。

func (*Dispatcher) SubmitWithOptions 

func (d *Dispatcher) SubmitWithOptions(ctx context.Context, name string, payload []byte, opts ...SubmitOption) (string, error)

SubmitWithOptions 提交任务并应用全部选项。

type EventHook 

type EventHook interface {
	// OnTaskEvent 在任务生命周期节点触发。
	OnTaskEvent(ctx context.Context, e TaskEvent)
}

EventHook 是可选事件钩子（默认 no-op），由 eventx 等外部适配器接入。

type Handler 

type Handler func(ctx context.Context, job Job) error

Handler 任务处理器：返回 nil 视为成功；返回错误触发重试策略； panic 由框架 recover 并按失败处理。

type Job 

type Job struct {
	// ID 全局唯一标识（32 位十六进制随机串）。
	ID string
	// Name 处理器路由名（非空且长度 ≤ 128）。
	Name string
	// Payload 载荷（业务自行编码，如 JSON）。
	Payload []byte
	// CreatedAt 创建时间。
	CreatedAt time.Time
	// RunAt 计划执行时间（零值表示立即执行）。
	RunAt time.Time
	// MaxRetries 失败重试次数上限（0 表示不重试）。
	MaxRetries int
	// RetryDelay 首次重试延迟（后续指数 ×2）。
	RetryDelay time.Duration
	// Timeout 单次执行超时（0 表示不限制）。
	Timeout time.Duration
	// Attempt 已尝试次数（框架维护，业务只读）。
	Attempt int
}

Job 任务单元。

type Metrics 

type Metrics struct {
	// Queued 就绪队列入队 +1 / 出队 -1。
	Queued func(name string, delta int)
	// Running 执行中 +1 / -1。
	Running func(name string, delta int)
	// Completed 任务成功完成。
	Completed func(name string, duration time.Duration)
	// Failed 任务最终失败。
	Failed func(name string, err error)
	// Retried 安排重试（attempt 为即将执行的第 N 次）。
	Retried func(name string, attempt int)
	// Dropped 关闭/队列满等策略丢弃。
	Dropped func(name string)
	// Skipped ConflictSkip 跳过。
	Skipped func(name string)
	// Replaced ConflictReplace 替换旧任务。
	Replaced func(name string)
}

Metrics 外部注入的任务指标回调（全部可选，nil 跳过）。

type Option 

type Option func(*config) error

Option Dispatcher 配置项。

func WithClock 

func WithClock(now func() time.Time) Option

WithClock 注入时间源（测试用）。

func WithConflictPolicy 

func WithConflictPolicy(p ConflictPolicy) Option

WithConflictPolicy 设置同名任务在途时的处理策略。

func WithEventHook 

func WithEventHook(h EventHook) Option

WithEventHook 设置任务事件钩子；不设置时 no-op。

func WithLogger 

func WithLogger(logger logx.Logger) Option

WithLogger 注入结构化日志器（nil 表示不记录）。

func WithMaxPayloadBytes 

func WithMaxPayloadBytes(n int) Option

WithMaxPayloadBytes 设置任务载荷长度上限（必须为正）。

func WithMetrics 

func WithMetrics(m Metrics) Option

WithMetrics 注入任务指标回调（全部可选）。

func WithQueueFullPolicy 

func WithQueueFullPolicy(p QueueFullPolicy) Option

WithQueueFullPolicy 设置队列满时的提交策略。

func WithQueueSize 

func WithQueueSize(n int) Option

WithQueueSize 设置就绪队列容量（必须为正）。

func WithStore 

func WithStore(store Store) Option

WithStore 启用任务持久化（提交/延迟同步写入，终态删除）。

func WithTraceHook 

func WithTraceHook(h TraceHook) Option

WithTraceHook 设置任务执行链路追踪钩子。

func WithWorkers 

func WithWorkers(n int) Option

WithWorkers 设置 worker 数量（必须为正）。

type QueueFullPolicy 

type QueueFullPolicy uint8

QueueFullPolicy 就绪队列满时的提交策略。 

const (
	// QueueFullBlock 阻塞提交直到队列有空间（默认，保证不丢任务）。
	QueueFullBlock QueueFullPolicy = iota
	// QueueFullDrop 丢弃新任务并返回 ErrQueueFull（不阻塞业务）。
	QueueFullDrop
)

type Schedule 

type Schedule struct {
	ID   string
	Name string
	Next time.Time
	// contains filtered or unexported fields
}

Schedule 调度条目快照。

func (*Schedule) Stop 

func (s *Schedule) Stop()

Stop 停止该调度条目（幂等）。

type Scheduler 

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

Scheduler 定时调度器：周期、一次性与 cron 表达式调度。 触发后提交给 Dispatcher 执行；所有方法并发安全。

func NewScheduler 

func NewScheduler(dispatcher *Dispatcher, opts ...SchedulerOption) (*Scheduler, error)

NewScheduler 构造调度器并绑定执行器。

func (*Scheduler) Cron 

func (s *Scheduler) Cron(expr, name string) (*Schedule, error)

Cron 注册 cron 表达式调度；表达式非法返回 ErrCronInvalid。

func (*Scheduler) DailyAt 

func (s *Scheduler) DailyAt(hour, minute, second int, name string) (*Schedule, error)

DailyAt 每天 hour:minute:second 触发（等价 cron 包装）。

func (*Scheduler) Every 

func (s *Scheduler) Every(interval time.Duration, name string) (*Schedule, error)

Every 注册周期调度，首次触发在第一个周期后。

func (*Scheduler) EveryHourAt 

func (s *Scheduler) EveryHourAt(minute int, name string) (*Schedule, error)

EveryHourAt 每小时第 minute 分 0 秒触发（等价 cron 包装）。

func (*Scheduler) EveryMinuteAt 

func (s *Scheduler) EveryMinuteAt(second int, name string) (*Schedule, error)

EveryMinuteAt 每分钟第 second 秒触发（等价 6 字段 cron 包装）。

func (*Scheduler) List 

func (s *Scheduler) List() []Schedule

List 返回全部调度条目快照。

func (*Scheduler) OneShot 

func (s *Scheduler) OneShot(at time.Time, name string) (*Schedule, error)

OneShot 注册一次性调度，触发后条目自动失效。

func (*Scheduler) Shutdown 

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

Shutdown 停止全部条目并等待退出；Dispatcher 由调用方关闭。

func (*Scheduler) Stop 

func (s *Scheduler) Stop(id string) error

Stop 停止并移除调度条目；不存在返回 ErrJobNotFound。

func (*Scheduler) WeeklyAt 

func (s *Scheduler) WeeklyAt(weekday, hour, minute, second int, name string) (*Schedule, error)

WeeklyAt 每周 weekday（0=周日）的 hour:minute:second 触发。

type SchedulerOption 

type SchedulerOption func(*schedulerConfig) error

SchedulerOption 调度器配置项。

func WithLocation 

func WithLocation(loc *time.Location) SchedulerOption

WithLocation 设置调度时区（默认本地时区）。

func WithSchedulerLogger 

func WithSchedulerLogger(logger logx.Logger) SchedulerOption

WithSchedulerLogger 注入调度日志器（nil 表示不记录）。

type Status 

type Status uint8

Status 任务状态。 

const (
	StatusQueued Status = iota
	StatusDelayed
	StatusRunning
	StatusSucceeded
	StatusFailed
	StatusCancelled
)

任务状态枚举。

type Store 

type Store interface {
	// Save 保存任务（新建或状态更新，如重试后的 Attempt/RunAt）。
	Save(ctx context.Context, job Job) error
	// Delete 删除任务（终态、取消或关闭丢弃）。
	Delete(ctx context.Context, id string) error
	// List 列出全部未完成任务（排队/延迟），供 Restore 恢复。
	List(ctx context.Context) ([]Job, error)
}

Store 任务持久化接口（可选）：将排队/延迟任务落库，进程重启后可恢复。 默认不启用（进程内任务随进程退出丢失）；启用后提交与延迟任务同步写入， 终态/取消同步删除。适配 Redis/数据库等外部实现由业务自行提供。

type SubmitOption 

type SubmitOption func(*jobSpec) error

SubmitOption 配置单次提交的任务。

func WithRetry 

func WithRetry(maxRetries int, delay time.Duration) SubmitOption

WithRetry 设置失败重试次数上限与首次重试延迟（后续指数 ×2）。

func WithRunAt 

func WithRunAt(at time.Time) SubmitOption

WithRunAt 设置计划执行时刻（零值报错；早于当前时刻按立即执行）。

func WithTimeout 

func WithTimeout(timeout time.Duration) SubmitOption

WithTimeout 设置单次执行超时（必须非负）。

type TaskEvent 

type TaskEvent struct {
	// Action 事件类型：queued / running / completed / failed / retried /
	// dropped / skipped / replaced。
	Action string
	// Name 任务名。
	Name string
	// Attempt 重试时的即将执行次数（仅 retried）。
	Attempt int
	// Err 失败原因（仅 failed）。
	Err error
}

TaskEvent 描述一次任务生命周期事件。

type TraceAttr 

type TraceAttr = contract.TraceAttr

TraceAttr 链路追踪属性（家族统一契约，定义见 tracex/contract）。

type TraceHook 

type TraceHook = contract.TraceHook

TraceHook 链路追踪钩子（家族统一契约，定义见 tracex/contract）。

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.