job

package
v1.7.4 Latest Latest
Warning

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

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

Documentation

Overview

Package job 提供长时间运行作业（Job）的状态机与执行能力。

与 queue 包（无状态消息投递）不同，job 包关注作业的完整生命周期： 提交（pending）→ 执行（running）→ 失败重试（pending + 调度时间）→ 终态（success/failed/cancelled）， 并提供进度上报、心跳保活、失联恢复与重试退避等长任务必需的能力。

重要语义：作业采用「至少一次执行」投递，即作业可能被重复执行， 处理器（Handler）必须实现幂等或显式容忍重复，并且尊重 ctx 的取消 （作业超时、取消或执行器停止时 ctx 会被关闭），否则会造成 goroutine 泄漏且作业无法及时终止。

取消延迟：对 running 作业调用 Cancel 只是更新库中状态，处理器要等到 下一次心跳（默认 HeartbeatInterval 10s）发现状态变化后才会收到 ctx 取消， 因此取消生效存在最多约一个心跳间隔的延迟。

处理器分发：Executor 支持按作业类型注册处理器（Register），未注册的 类型回退到 NewExecutor 传入的默认处理器；两者都没有时会按重试策略 重新入队，超过 MaxAttempts 后标记失败。

存储仅支持 MySQL 5.7+：PopPending 使用「无锁 SELECT + 原子 UPDATE 抢占」 （UPDATE 携带 status/attempts 条件 + RowsAffected 判定），锁等待/死锁自动重试； 数据库与连接需使用 utf8mb4 字符集（避免中文/emoji 乱码）。 Payload/Result 存储在 MySQL JSON 列中，写入时校验 JSON 合法性， 提交作业时应传入合法 JSON（如 []byte(`{"key":"value"}`)）， 便于在数据库中通过 JSON_EXTRACT 等函数查询任务内容。

Index

Constants

View Source 
const DefaultQueue = "default"

DefaultQueue 默认队列名

Variables

View Source 
var (
	// ErrJobNotFound 作业不存在
	ErrJobNotFound = errors.New("job not found")
	// ErrNilJob 作业为空
	ErrNilJob = errors.New("job cannot be nil")
	// ErrEmptyJobType 作业类型为空
	ErrEmptyJobType = errors.New("job type cannot be empty")
	// ErrJobStateConflict 作业状态与预期不符，状态转移未生效
	ErrJobStateConflict = errors.New("job state conflict")
)

作业相关公共错误

View Source 
var ErrExecutorRunning = errors.New("job: executor is already running")

ErrExecutorRunning 执行器已启动

View Source 
var ErrStopTimeout = errors.New("job: executor stop timed out")

ErrStopTimeout 执行器停止超时，仍有处理器未退出

Functions

func ReportProgress 

func ReportProgress(ctx context.Context, progress int)

ReportProgress 从上下文中上报作业执行进度（0-100），超出范围会被截断。 只有在执行器创建的上下文中调用才生效，否则为无操作。

Types

type AttemptError 

type AttemptError struct {
	Attempt int    `json:"attempt"` // 尝试次数（1-based）
	At      int64  `json:"at"`      // 失败时间（Unix 毫秒）
	Error   string `json:"error"`   // 错误信息
}

AttemptError 记录单次执行尝试的错误信息

type Config 

type Config struct {
	Queues            []string      // 要消费的队列名
	Concurrency       int           // 每队列并发执行数
	HeartbeatInterval time.Duration // 心跳间隔
	StaleTimeout      time.Duration // 心跳超时判定（超过则视为失联，重新入队）
	RecoverInterval   time.Duration // 失联作业扫描间隔
	JobTimeout        time.Duration // 单个作业执行超时（0 表示不限制）
	RetryDelay        time.Duration // 重试退避基础延迟
	RetryMaxDelay     time.Duration // 重试退避最大延迟
	RetryStrategy     RetryStrategy // 重试退避策略
	RetryJitter       bool          // 重试延迟加入随机抖动（默认开启，避免惊群）
	Retention         time.Duration // 终态作业保留时长（0 表示永久保留，不自动清理）
	CleanupInterval   time.Duration // 终态作业清理间隔（Retention > 0 时生效）
	ShutdownTimeout   time.Duration // 优雅停止等待处理器退出的超时（0 使用默认 30s）
	ProgressInterval  time.Duration // 进度上报节流间隔（合并高频上报，0 表示不节流）
}

Config 作业执行器配置

func DefaultConfig 

func DefaultConfig() Config

DefaultConfig 返回默认配置

func (Config) Validate 

func (c Config) Validate() error

Validate 校验配置合法性，返回首个不合法的配置项错误。 零值或负值的周期配置会导致 ticker panic 或失联判定失效，因此必须在启动前校验。

func (Config) WithCleanupInterval 

func (c Config) WithCleanupInterval(interval time.Duration) Config

WithCleanupInterval 设置终态作业清理间隔

func (Config) WithConcurrency 

func (c Config) WithConcurrency(n int) Config

WithConcurrency 设置每队列并发执行数

func (Config) WithHeartbeatInterval 

func (c Config) WithHeartbeatInterval(interval time.Duration) Config

WithHeartbeatInterval 设置心跳间隔

func (Config) WithJobTimeout 

func (c Config) WithJobTimeout(timeout time.Duration) Config

WithJobTimeout 设置单个作业执行超时（0 表示不限制）

func (Config) WithProgressInterval 

func (c Config) WithProgressInterval(interval time.Duration) Config

WithProgressInterval 设置进度上报节流间隔（0 表示不节流，每次上报立即落库）

func (Config) WithQueues 

func (c Config) WithQueues(queues ...string) Config

WithQueues 设置要消费的队列名

func (Config) WithRecoverInterval 

func (c Config) WithRecoverInterval(interval time.Duration) Config

WithRecoverInterval 设置失联扫描间隔

func (Config) WithRetention 

func (c Config) WithRetention(d time.Duration) Config

WithRetention 设置终态作业保留时长（0 表示永久保留）

func (Config) WithRetryDelay 

func (c Config) WithRetryDelay(delay time.Duration) Config

WithRetryDelay 设置重试退避基础延迟

func (Config) WithRetryJitter 

func (c Config) WithRetryJitter(jitter bool) Config

WithRetryJitter 设置重试延迟是否加入随机抖动（默认开启，避免惊群）

func (Config) WithRetryMaxDelay 

func (c Config) WithRetryMaxDelay(delay time.Duration) Config

WithRetryMaxDelay 设置重试退避最大延迟

func (Config) WithRetryStrategy 

func (c Config) WithRetryStrategy(strategy RetryStrategy) Config

WithRetryStrategy 设置重试退避策略

func (Config) WithShutdownTimeout 

func (c Config) WithShutdownTimeout(timeout time.Duration) Config

WithShutdownTimeout 设置优雅停止等待处理器退出的超时（0 使用默认 30s）

func (Config) WithStaleTimeout 

func (c Config) WithStaleTimeout(timeout time.Duration) Config

WithStaleTimeout 设置心跳超时判定

type Executor 

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

Executor 消费 pending 作业并执行，负责心跳保活、进度上报、失败重试与失联恢复。

func NewExecutor 

func NewExecutor(store Store, handler Handler, config Config) *Executor

NewExecutor 创建作业执行器。 handler 作为默认处理器：未通过 Register 注册的类型会回退到它； 若所有类型都通过 Register 注册，handler 可传 nil。

func (*Executor) IsRunning 

func (e *Executor) IsRunning() bool

IsRunning 报告执行器是否正在运行

func (*Executor) Register 

func (e *Executor) Register(jobType string, handler Handler) *Executor

Register 为指定作业类型注册处理器，需在 Start 之前调用。 空类型或 nil 处理器会被忽略。未注册的类型回退到 NewExecutor 传入的默认处理器。

func (*Executor) Start 

func (e *Executor) Start(ctx context.Context) error

Start 启动执行器，为每个队列创建 config.Concurrency 个执行协程及维护协程

func (*Executor) Stop 

func (e *Executor) Stop() error

Stop 优雅停止执行器，取消正在执行的作业并等待退出。 若处理器未在 ShutdownTimeout 内响应取消，则强制返回 ErrStopTimeout， 残留的 running 作业将由下次启动的失联恢复接管。

func (*Executor) WithLogger 

func (e *Executor) WithLogger(log logger.Logger) *Executor

WithLogger 设置执行器的日志实例，需在 Start 之前调用。

type Failure 

type Failure struct {
	Error  string         // 最近一次错误信息
	Errors []AttemptError // 完整错误历史
}

Failure 作业执行失败信息

type Handler 

type Handler interface {
	// Execute 执行作业，返回 error 表示执行失败。
	//
	// 执行结果通过修改 job.Result 返回；
	// 执行进度通过 ReportProgress(ctx, progress) 上报。
	// ctx 在作业被取消、超时或执行器停止时会被关闭，处理器应尊重 ctx 的取消。
	Execute(ctx context.Context, job *Job) error
}

Handler 作业处理器接口

type HandlerFunc 

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

HandlerFunc 函数类型适配 Handler 接口

func (HandlerFunc) Execute 

func (f HandlerFunc) Execute(ctx context.Context, job *Job) error

Execute 实现 Handler 接口

type Job 

type Job struct {
	ID          string            `json:"id"`           // 作业ID
	Queue       string            `json:"queue"`        // 队列名
	Type        string            `json:"type"`         // 作业类型
	Payload     []byte            `json:"payload"`      // 作业输入数据（JSON）
	Status      Status            `json:"status"`       // 作业状态
	Progress    int               `json:"progress"`     // 执行进度（0-100）
	Result      []byte            `json:"result"`       // 作业输出数据（JSON）
	Error       string            `json:"error"`        // 最近一次错误信息
	Attempts    int               `json:"attempts"`     // 已执行次数（1-based，首次执行为 1）
	MaxAttempts int               `json:"max_attempts"` // 最大执行次数（默认 1，即不重试）
	ScheduledAt int64             `json:"scheduled_at"` // 下次可执行时间（Unix 毫秒，0 表示立即）
	CreatedAt   int64             `json:"created_at"`   // 创建时间（Unix 毫秒）
	StartedAt   int64             `json:"started_at"`   // 最近开始执行时间（Unix 毫秒）
	CompletedAt int64             `json:"completed_at"` // 完成时间（Unix 毫秒）
	HeartbeatAt int64             `json:"heartbeat_at"` // 最近心跳时间（Unix 毫秒）
	Metadata    map[string]string `json:"metadata"`     // 元数据
	Errors      []AttemptError    `json:"errors"`       // 错误历史
}

Job 表示一个长时间运行的作业

func NewJob 

func NewJob(jobType string, payload []byte) *Job

NewJob 创建新作业

func (*Job) IsTerminal 

func (j *Job) IsTerminal() bool

IsTerminal 报告作业是否处于终态（成功/失败/取消）

func (*Job) WithDelay 

func (j *Job) WithDelay(delay time.Duration) *Job

WithDelay 设置延迟执行（从当前时间起算）

func (*Job) WithMaxAttempts 

func (j *Job) WithMaxAttempts(maxAttempts int) *Job

WithMaxAttempts 设置最大执行次数（含首次，默认 1 表示不重试）

func (*Job) WithMetadata 

func (j *Job) WithMetadata(key, value string) *Job

WithMetadata 设置元数据

func (*Job) WithPayload 

func (j *Job) WithPayload(payload []byte) *Job

WithPayload 设置作业输入数据

func (*Job) WithQueue 

func (j *Job) WithQueue(queue string) *Job

WithQueue 设置队列名

func (*Job) WithScheduleAt 

func (j *Job) WithScheduleAt(t time.Time) *Job

WithScheduleAt 设置定时执行时间

type JobFilter 

type JobFilter struct {
	Queue  string    // 队列过滤（空表示全部）
	Type   string    // 类型过滤（空表示全部）
	Status Status    // 状态过滤（空表示全部）
	Since  time.Time // 创建时间下限（零值表示不限）
	Until  time.Time // 创建时间上限（零值表示不限）
	Limit  int       // 返回数量上限（<=0 使用默认值 100）
	Offset int       // 偏移量
}

JobFilter 作业查询过滤条件，零值字段表示不限制

type RetryStrategy 

type RetryStrategy string

RetryStrategy 重试退避策略 

const (
	// RetryStrategyFixed 固定延迟
	RetryStrategyFixed RetryStrategy = "fixed"
	// RetryStrategyExponential 指数退避
	RetryStrategyExponential RetryStrategy = "exponential"
)

type SQLStore 

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

SQLStore 基于 MySQL 的作业存储实现，通过 gorm 适配。 仅支持 MySQL 5.7+，库与连接需使用 utf8mb4 字符集。

func NewSQLStore 

func NewSQLStore(db *gorm.DB, opts ...SQLStoreOption) (*SQLStore, error)

NewSQLStore 创建 MySQL 作业存储实例并自动迁移表结构。 db 由调用方通过项目的 mysql 包创建并管理生命周期，需保证数据库字符集为 utf8mb4。 可通过 WithTableName 等选项自定义表名。

func (*SQLStore) Cancel 

func (s *SQLStore) Cancel(ctx context.Context, queue, jobID string) error

Cancel 取消作业（pending/running → cancelled）。 幂等：作业已处于终态（success/failed/cancelled）时返回 nil；作业不存在时返回 ErrJobNotFound。

func (*SQLStore) Cleanup 

func (s *SQLStore) Cleanup(ctx context.Context, retainFor time.Duration) (int64, error)

Cleanup 清理超过 retainFor 时长的终态作业

func (*SQLStore) Complete 

func (s *SQLStore) Complete(ctx context.Context, queue, jobID string, attempt int, result []byte) error

Complete 标记作业成功

func (*SQLStore) Delete 

func (s *SQLStore) Delete(ctx context.Context, jobID string) error

Delete 删除作业

func (*SQLStore) Fail 

func (s *SQLStore) Fail(ctx context.Context, queue, jobID string, attempt int, failure Failure) error

Fail 标记作业最终失败

func (*SQLStore) Get 

func (s *SQLStore) Get(ctx context.Context, jobID string) (*Job, error)

Get 获取作业

func (*SQLStore) GetStats 

func (s *SQLStore) GetStats(ctx context.Context, queue string) (*Stats, error)

GetStats 获取队列统计信息（按当前存在的各状态作业数）

func (*SQLStore) Heartbeat 

func (s *SQLStore) Heartbeat(ctx context.Context, queue, jobID string, attempt, progress int) (bool, error)

Heartbeat 更新心跳与进度，返回是否仍处于 running 状态

func (*SQLStore) List 

func (s *SQLStore) List(ctx context.Context, filter JobFilter) ([]*Job, error)

List 按过滤条件查询作业，结果按创建时间降序排列

func (*SQLStore) ListStaleRunning 

func (s *SQLStore) ListStaleRunning(ctx context.Context, queue string, since time.Time, limit int64) ([]*Job, error)

ListStaleRunning 列出心跳早于 since 的 running 作业

func (*SQLStore) PopPending 

func (s *SQLStore) PopPending(ctx context.Context, queue string) (*Job, error)

PopPending 原子取出最早的一个 pending 作业并标记 running

func (*SQLStore) Requeue 

func (s *SQLStore) Requeue(ctx context.Context, queue, jobID string) error

Requeue 将失联作业重新入队（running → pending，立即可执行）

func (*SQLStore) Retry 

func (s *SQLStore) Retry(ctx context.Context, queue, jobID string, attempt int, retryAt time.Time, failure Failure) error

Retry 将失败的 running 作业重新入队为 pending，并将下次可执行时间设为 retryAt

func (*SQLStore) Save 

func (s *SQLStore) Save(ctx context.Context, job *Job) error

Save 创建作业并按调度时间设置初始状态。 不修改传入的 job：默认队列、重试次数与初始状态仅作用于落库数据，调用方应通过 Get 读取最终值。

type SQLStoreOption 

type SQLStoreOption func(*SQLStore)

SQLStoreOption 配置 SQLStore 的选项

func WithTableName 

func WithTableName(name string) SQLStoreOption

WithTableName 设置作业表名（默认 "jobs"），用于多服务共库或表名前缀等场景

type Stats 

type Stats struct {
	Pending   int64 `json:"pending"`   // 等待执行数（含延迟/重试等待）
	Running   int64 `json:"running"`   // 执行中数
	Success   int64 `json:"success"`   // 成功状态数
	Failed    int64 `json:"failed"`    // 失败状态数
	Cancelled int64 `json:"cancelled"` // 取消状态数
}

Stats 队列统计信息（各状态当前存在的作业数）

type Status 

type Status string

Status 作业状态 

const (
	// StatusPending 等待执行（含延迟/重试等待，由 ScheduledAt 控制实际可执行时间）
	StatusPending Status = "pending"
	// StatusRunning 执行中
	StatusRunning Status = "running"
	// StatusSuccess 执行成功
	StatusSuccess Status = "success"
	// StatusFailed 执行失败
	StatusFailed Status = "failed"
	// StatusCancelled 已取消
	StatusCancelled Status = "cancelled"
)

type Store 

type Store interface {
	// Save 保存（创建）作业，按 job.Queue 与 job.ScheduledAt 决定入队位置
	Save(ctx context.Context, job *Job) error

	// Get 获取作业
	Get(ctx context.Context, jobID string) (*Job, error)

	// Delete 删除作业及其索引
	Delete(ctx context.Context, jobID string) error

	// PopPending 原子取出队列中最早到期的一个 pending 作业（scheduled_at <= now）并标记 running，
	// 无到期作业时返回 (nil, nil)。
	PopPending(ctx context.Context, queue string) (*Job, error)

	// Heartbeat 更新 running 作业的心跳与进度，返回作业是否仍处于 running 状态。
	// progress < 0 表示不更新进度，仅刷新心跳。
	// attempt 为当前执行尝试（乐观锁版本号），与库中不匹配时返回 running=false。
	Heartbeat(ctx context.Context, queue, jobID string, attempt, progress int) (bool, error)

	// Complete 标记 running 作业为 success 并写入结果。
	// 作业不在 running 状态或 attempt 不匹配（已被重新调度）时返回 ErrJobStateConflict。
	Complete(ctx context.Context, queue, jobID string, attempt int, result []byte) error

	// Fail 标记 running 作业为 failed（最终失败），并记录错误信息。
	// 作业不在 running 状态或 attempt 不匹配时返回 ErrJobStateConflict。
	Fail(ctx context.Context, queue, jobID string, attempt int, failure Failure) error

	// Retry 将失败的 running 作业重新入队为 pending，并将下次可执行时间设为 retryAt。
	// 作业不在 running 状态或 attempt 不匹配时返回 ErrJobStateConflict。
	Retry(ctx context.Context, queue, jobID string, attempt int, retryAt time.Time, failure Failure) error

	// Cancel 取消作业（pending/running → cancelled）。
	// 幂等：作业已处于终态时返回 nil；作业不存在时返回 ErrJobNotFound。
	Cancel(ctx context.Context, queue, jobID string) error

	// Requeue 将失联的 running 作业重新入队（running → pending，立即可执行）。
	// 作业不在 running 状态时返回 ErrJobStateConflict。
	Requeue(ctx context.Context, queue, jobID string) error

	// ListStaleRunning 列出心跳早于 since 的 running 作业（用于失联恢复）
	ListStaleRunning(ctx context.Context, queue string, since time.Time, limit int64) ([]*Job, error)

	// GetStats 获取队列统计信息
	GetStats(ctx context.Context, queue string) (*Stats, error)

	// List 按过滤条件查询作业，结果按创建时间降序排列
	List(ctx context.Context, filter JobFilter) ([]*Job, error)

	// Cleanup 清理超过 retainFor 时长的终态作业（success/failed/cancelled），返回清理数量
	Cleanup(ctx context.Context, retainFor time.Duration) (int64, error)
}

Store 作业存储接口，负责作业状态的持久化与原子状态转移。

实现必须保证状态转移（如 PopPending、Complete、Fail、Retry）的原子性， 以避免并发执行器重复消费同一作业。

作业详情按全局唯一 ID 存储，队列索引按 queue 维度隔离。

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.