任务创建与入队

func NewTask(typename string, payload []byte, opts ...Option) *Task

// 基础创建
task := asynq.NewTask("email:send", payload)

// 带选项创建（选项也可以在 Enqueue 时传递，效果相同）
task := asynq.NewTask("email:send", payload,
    asynq.Queue("critical"),
    asynq.MaxRetry(5),
)

// ✅ 推荐：简洁稳定的结构
type ExportPayload struct {
    ReportID  string `json:"report_id"`
    Format    string `json:"format"`    // "pdf" | "csv"
    NotifyURL string `json:"notify_url"` // 完成后回调
}

// ❌ 避免：Payload 中嵌入大段数据
type BadPayload struct {
    ImageData []byte `json:"image_data"` // 应该存对象存储传 URL
}

client := asynq.NewClient(asynq.RedisClientOpt{Addr: "localhost:6379"})
defer client.Close() // 确保释放 Redis 连接

// 30 分钟后执行
client.Enqueue(task, asynq.ProcessIn(30*time.Minute))

// 明天上午 9 点执行
tomorrow9am := time.Now().Add(24*time.Hour).Truncate(24*time.Hour).Add(9*time.Hour)
client.Enqueue(task, asynq.ProcessAt(tomorrow9am))

// ProcessIn 和 ProcessAt 互斥，同时使用以最后一个为准

// 单次执行超过 5 分钟则强制终止（context 取消）
client.Enqueue(task, asynq.Timeout(5*time.Minute))

// 超过绝对 deadline 后任务不再被调度
client.Enqueue(task, asynq.Deadline(time.Now().Add(1*time.Hour)))

info, err := client.Enqueue(task,
    asynq.Queue("critical"),
    asynq.ProcessIn(10*time.Minute),
    asynq.MaxRetry(3),
    asynq.Timeout(30*time.Second),
    asynq.Deadline(time.Now().Add(2*time.Hour)),
    asynq.Unique(1*time.Hour),
    asynq.Retention(24*time.Hour),
)

client.Enqueue(task, asynq.Unique(10*time.Minute))

// ✅ 推荐：Payload 中只包含稳定的业务标识
type DedupPayload struct {
    OrderID  int    `json:"order_id"`   // 业务唯一标识
    Action   string `json:"action"`     // "cancel" | "refund"
}

// ❌ 错误：包含时间戳导致去重失效
type BadPayload struct {
    OrderID   int   `json:"order_id"`
    CreatedAt int64 `json:"created_at"` // 每次不同，去重无效
}

client.Enqueue(task, asynq.TaskID("order-12345-cancel"))

tasks := []*asynq.Task{task1, task2, task3}
infos, err := client.EnqueueBatch(tasks)

task := asynq.NewTask("my:type", payload,
    asynq.Headers(map[string]string{
        "X-Trace-ID":    traceID,
        "X-Tenant-ID":   tenantID,
        "X-Request-ID":  requestID,
    }),
)

// Handler 中读取
func Handler(ctx context.Context, t *asynq.Task) error {
    traceID := t.Headers["X-Trace-ID"]
    // ...
}

// 方式 1：NewTask 时指定
task := asynq.NewTask("email:send", payload, asynq.Queue("critical"))

// 方式 2：Enqueue 时指定（覆盖 NewTask 时的设置）
client.Enqueue(task, asynq.Queue("critical"))

// 任务完成后在 Redis 中保留 24 小时，便于查询和审计
client.Enqueue(task, asynq.Retention(24*time.Hour))

info, err := client.Enqueue(task)
if err != nil {
    if errors.Is(err, asynq.ErrTaskIdConflict) {
        // 自定义 TaskID 冲突
    }
    log.Printf("enqueue failed: %v", err)
    return
}
log.Printf("task enqueued: id=%s queue=%s state=%s", info.ID, info.Queue, info.State)

package tasks

import (
    "encoding/json"
    "fmt"

    "github.com/hibiken/asynq"
)

// ─── 类型常量 ───────────────────────

const (
    TypeEmailSend     = "email:send"
    TypeOrderCancel   = "order:cancel"
    TypeReportExport  = "report:export"
    TypeImageProcess  = "image:process"
)

// ─── Payload 定义 ───────────────────

type EmailSendPayload struct {
    To      string `json:"to"`
    Subject string `json:"subject"`
    Body    string `json:"body"`
}

type OrderCancelPayload struct {
    OrderID int64  `json:"order_id"`
    Reason  string `json:"reason"`
}

type ReportExportPayload struct {
    ReportID   string `json:"report_id"`
    Format     string `json:"format"`      // "pdf" | "csv" | "xlsx"
    CallbackURL string `json:"callback_url"` // 完成后通知
}

type ImageProcessPayload struct {
    ImageURL   string `json:"image_url"`
    Operations []string `json:"operations"` // "resize", "watermark", "compress"
}

// ─── 任务构造函数 ──────────────────

func NewEmailSendTask(p EmailSendPayload) (*asynq.Task, error) {
    data, err := json.Marshal(p)
    if err != nil {
        return nil, fmt.Errorf("marshal: %w", err)
    }
    return asynq.NewTask(TypeEmailSend, data,
        asynq.Queue("default"),
        asynq.MaxRetry(3),
        asynq.Retention(24*time.Hour),
    ), nil
}

func NewOrderCancelTask(orderID int64, reason string) (*asynq.Task, error) {
    data, err := json.Marshal(OrderCancelPayload{orderID, reason})
    if err != nil {
        return nil, err
    }
    return asynq.NewTask(TypeOrderCancel, data,
        asynq.Queue("critical"),
        asynq.MaxRetry(5),
        asynq.TaskID(fmt.Sprintf("order:%d:cancel", orderID)), // 业务去重
    ), nil
}