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 GetTaskID(ctx context.Context) string
• func NewRedisConnOpt() asynq.RedisConnOpt
• func PrefixedQueue(queue string) string
• func ProcessTask(ctx context.Context, t *asynq.Task) error
• func RegisterHandler(asynqTaskType string, handler TaskHandler)
• func RetryTask(ctx context.Context, id uint64) (string, error)
• func ValidateAndNormalizePayload(asynqTaskType string, payload []byte) ([]byte, error)
• type PayloadValidator
• type TaskHandler
• type TaskMeta
  • func GetTaskMeta(taskType string) *TaskMeta
• type TaskParam
• type TaskResult

Constants

const (
	CleanupUnusedUploadsTask = "upload:cleanup_unused"
	SendEmailTask            = "mail:send"
)

异步任务类型标识

const (
	TaskTypeCleanupUploads = "cleanup_unused_uploads"
	TaskTypeSendEmail      = "send_email"
)

管理员可下发的任务类型标识

const (
	QueueDefault = "default"
)

任务队列名称

Variables

var AsynqClient *asynq.Client

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

var DispatchableTasks = []TaskMeta{
	{
		Type:         TaskTypeCleanupUploads,
		AsynqTask:    CleanupUnusedUploadsTask,
		Name:         "清理未使用上传",
		Description:  "清理超过1小时未使用的上传文件",
		SupportsTime: false,
		MaxRetry:     defaultMaxRetry,
		Queue:        QueueDefault,
		Retryable:    true,
	},
	{
		Type:         TaskTypeSendEmail,
		AsynqTask:    SendEmailTask,
		Name:         "发送邮件",
		Description:  "异步发送系统邮件",
		SupportsTime: false,
		MaxRetry:     defaultMaxRetry,
		Queue:        QueueDefault,
		Retryable:    true,
		Params: []TaskParam{
			{
				Name:        "to",
				Label:       "接收邮箱 (To)",
				Type:        "string",
				Required:    true,
				Placeholder: "receiver@example.com",
				Description: "接收邮件的目标邮箱地址",
			},
			{
				Name:        "subject",
				Label:       "邮件主题 (Subject)",
				Type:        "string",
				Required:    true,
				Placeholder: "请输入邮件主题",
				Description: "发送邮件的主题标题",
			},
			{
				Name:        "body",
				Label:       "邮件内容 (Body)",
				Type:        "text",
				Required:    true,
				Placeholder: "请输入邮件内容（支持 HTML 格式）",
				Description: "发送邮件的内容主体",
			},
		},
	},
}

DispatchableTasks 可下发的任务列表

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 GetTaskID

func GetTaskID(ctx context.Context) string

GetTaskID 从 context 中获取 taskID

func NewRedisConnOpt

func NewRedisConnOpt() asynq.RedisConnOpt

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

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 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 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
	AsynqTask    string
	Name         string
	Description  string
	SupportsTime bool
	MaxRetry     int
	Queue        string
	Retryable    bool // 是否支持手动重试
	Params       []TaskParam
}

TaskMeta 任务元数据

func GetTaskMeta

func GetTaskMeta(taskType string) *TaskMeta

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

type TaskParam

type TaskParam struct {
	Name        string `json:"Name"`        // 参数键名
	Label       string `json:"Label"`       // 显示名称
	Type        string `json:"Type"`        // 类型：string, text, number
	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 任务执行结果