task

Documentation

Overview

Package task 定义异步任务类型与调度常量

Index

• Constants
• Variables
• func AppendLog(ctx context.Context, format string, args ...interface{})
• func DispatchTask(ctx context.Context, taskType string, payload []byte, triggeredBy string) (string, error)
• func GetRegisteredAsynqTasks() []string
• func GetTaskID(ctx context.Context) string
• func IsFinalAttempt(ctx context.Context) bool
• func NewRedisConnOpt() asynq.RedisConnOpt
• func OnTaskCompleted(handler CompletedHandler)
• func PrefixedQueue(queue string) string
• func ProcessTask(ctx context.Context, t *asynq.Task) error
• func RegisterHandler(asynqTaskType string, handler TaskHandler)
• func RegisterTaskMeta(meta TaskMeta)
• func RetryTask(ctx context.Context, id uint64) (string, error)
• func ValidateAndNormalizePayload(asynqTaskType string, payload []byte) ([]byte, error)
• type CompletedHandler
• type PayloadValidator
• type TaskHandler
• type TaskMeta
  • func GetDispatchableTasks() []TaskMeta
  • func GetTaskMeta(taskType string) *TaskMeta
  • func GetTaskMetaByAsynqTask(asynqTask string) *TaskMeta
• type TaskParam
• type TaskResult

Constants

const DefaultMaxRetry = 3

DefaultMaxRetry 任务默认最大重试次数

const (
	QueueDefault = "default"
)

任务队列名称

Variables

var AsynqClient *asynq.Client

AsynqClient asynq 客户端，用于任务入队

var RedisOpt asynq.RedisConnOpt

RedisOpt asynq Redis 连接配置（兼容 Standalone/Sentinel/Cluster）

Functions

func AppendLog

func AppendLog(ctx context.Context, format string, args ...interface{})

AppendLog 追加日志到任务执行记录 在 TaskHandler.Execute 中调用，日志会自动追加到 TaskExecution.Log 字段

func DispatchTask

func DispatchTask(ctx context.Context, taskType string, payload []byte, triggeredBy string) (string, error)

DispatchTask 下发任务（创建 TaskExecution 记录 → 入队 Asynq）

func GetRegisteredAsynqTasks

func GetRegisteredAsynqTasks() []string

GetRegisteredAsynqTasks 返回所有已注册的 Asynq 任务名称，以便动态注册路由

func GetTaskID

func GetTaskID(ctx context.Context) string

GetTaskID 从 context 中获取 taskID

func IsFinalAttempt

func IsFinalAttempt(ctx context.Context) bool

IsFinalAttempt 判断当前任务执行是否为最后一次重试尝试（若再次失败即为最终失败）

func NewRedisConnOpt

func NewRedisConnOpt() asynq.RedisConnOpt

NewRedisConnOpt 根据配置返回对应的 asynq Redis 连接选项

func OnTaskCompleted

func OnTaskCompleted(handler CompletedHandler)

OnTaskCompleted registers a handler for task completion events. Handlers must be registered during application bootstrap before processing tasks.

func PrefixedQueue

func PrefixedQueue(queue string) string

PrefixedQueue 返回带前缀的队列名，用于 Cluster 模式隔离

func ProcessTask

func ProcessTask(ctx context.Context, t *asynq.Task) error

ProcessTask Asynq 实际调用的统一处理函数 Worker 注册时统一使用此函数，内部自动分发到对应的 TaskHandler

func RegisterHandler

func RegisterHandler(asynqTaskType string, handler TaskHandler)

RegisterHandler 注册任务处理器 传入任务类型标识（对应 constants.go 中的 AsynqTask 常量）和 TaskHandler 实现

func RegisterTaskMeta

func RegisterTaskMeta(meta TaskMeta)

RegisterTaskMeta 注册任务元数据到全局列表

func RetryTask

func RetryTask(ctx context.Context, id uint64) (string, error)

RetryTask 重试失败的任务

func ValidateAndNormalizePayload

func ValidateAndNormalizePayload(asynqTaskType string, payload []byte) ([]byte, error)

ValidateAndNormalizePayload 校验并标准化任务参数。 如果 Handler 实现了 PayloadValidator，调用其 ValidatePayload 方法； 否则直接返回原始 payload。

Types

type CompletedHandler

type CompletedHandler func(ctx context.Context, execution *model.TaskExecution, result *TaskResult, execErr error)

CompletedHandler is called when a task execution completes.

type PayloadValidator

type PayloadValidator interface {
	ValidatePayload(payload []byte) ([]byte, error)
}

PayloadValidator 可选接口，带参数的任务 Handler 应实现此接口。 框架在 Admin 下发时自动调用，完成参数校验和标准化（如 Trim 空白）。 无参数的任务无需实现，框架会直接透传 payload。

type TaskHandler

type TaskHandler interface {
	// Execute 执行任务业务逻辑
	//   - ctx: 已注入 Trace Span 和 taskID 的上下文
	//   - payload: 调度时传入的原始参数（可为 nil）
	//   - 返回 TaskResult 描述执行结果，或 error 表示执行失败
	Execute(ctx context.Context, payload []byte) (*TaskResult, error)
}

TaskHandler 异步任务处理器接口 所有异步任务必须实现此接口，框架将自动管理任务执行记录的创建、状态流转和日志写入。

开发者只需实现 Execute 方法编写业务逻辑，在方法内通过 task.AppendLog(ctx, ...) 追加执行日志。 任务的创建、状态更新、错误记录、重试计数全部由框架透明处理。

type TaskMeta

type TaskMeta struct {
	Type         string      `json:"type"`
	AsynqTask    string      `json:"asynq_task"`
	Name         string      `json:"name"`
	Description  string      `json:"description"`
	SupportsTime bool        `json:"supports_time"`
	MaxRetry     int         `json:"max_retry"`
	Queue        string      `json:"queue"`
	Retryable    bool        `json:"retryable"` // 是否支持手动重试
	Params       []TaskParam `json:"params,omitempty"`
}

TaskMeta 任务元数据

func GetDispatchableTasks

func GetDispatchableTasks() []TaskMeta

GetDispatchableTasks 获取所有已注册的元数据列表（返回副本以避免并发并发读写冲突）

func GetTaskMeta

func GetTaskMeta(taskType string) *TaskMeta

GetTaskMeta 根据任务类型获取元数据

func GetTaskMetaByAsynqTask

func GetTaskMetaByAsynqTask(asynqTask string) *TaskMeta

GetTaskMetaByAsynqTask 根据 Asynq 任务名称获取元数据

type TaskParam

type TaskParam struct {
	Name        string `json:"name"`        // 参数键名
	Label       string `json:"label"`       // 显示名称
	Type        string `json:"type"`        // 类型：string, text, number, boolean
	Required    bool   `json:"required"`    // 是否必填
	Placeholder string `json:"placeholder"` // 占位符
	Description string `json:"description"` // 描述
}

TaskParam 任务参数定义

type TaskResult

type TaskResult struct {
	Message string // 结果摘要，如 "共清理 120 个文件，耗时 3.2s"
	Detail  string // 可选的详细结果 JSON
}

TaskResult 任务执行结果