简单、可靠且高效的Go分布式任务队列

package tasks

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"
    "github.com/hibiken/asynq"
)

// 任务类型列表
const (
    TypeEmailDelivery   = "email:deliver"
    TypeImageResize     = "image:resize"
)

type EmailDeliveryPayload struct {
    UserID     int
    TemplateID string
}

type ImageResizePayload struct {
    SourceURL string
}

//----------------------------------------------
// 编写一个NewXXXTask函数来创建任务。
// 任务由类型和有效负载组成。
//----------------------------------------------

func NewEmailDeliveryTask(userID int, tmplID string) (*asynq.Task, error) {
    payload, err := json.Marshal(EmailDeliveryPayload{UserID: userID, TemplateID: tmplID})
    if err != nil {
        return nil, err
    }
    return asynq.NewTask(TypeEmailDelivery, payload), nil
}

func NewImageResizeTask(src string) (*asynq.Task, error) {
    payload, err := json.Marshal(ImageResizePayload{SourceURL: src})
    if err != nil {
        return nil, err
    }
    // 可以将任务选项传递给NewTask,这些选项可以在入队时被覆盖。
    return asynq.NewTask(TypeImageResize, payload, asynq.MaxRetry(5), asynq.Timeout(20 * time.Minute)), nil
}

//---------------------------------------------------------------
// 编写一个HandleXXXTask函数来处理输入任务。
// 注意它满足asynq.HandlerFunc接口。
//
// 处理程序不需要是函数。您可以定义一个满足asynq.Handler接口的类型。
// 请参见下面的示例。
//---------------------------------------------------------------

func HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
    var p EmailDeliveryPayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("发送电子邮件给用户: user_id=%d, template_id=%s", p.UserID, p.TemplateID)
    // 电子邮件发送代码 ...
    return nil
}

// ImageProcessor实现asynq.Handler接口。
type ImageProcessor struct {
    // ... 结构体的字段
}

func (processor *ImageProcessor) ProcessTask(ctx context.Context, t *asynq.Task) error {
    var p ImageResizePayload
    if err := json.Unmarshal(t.Payload(), &p); err != nil {
        return fmt.Errorf("json.Unmarshal failed: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("调整图像大小: src=%s", p.SourceURL)
    // 图像调整大小代码 ...
    return nil
}

func NewImageProcessor() *ImageProcessor {
	return &ImageProcessor{}
}

package main

import (
    "log"
    "time"

    "github.com/hibiken/asynq"
    "your/app/package/tasks"
)

const redisAddr = "127.0.0.1:6379"

func main() {
    client := asynq.NewClient(asynq.RedisClientOpt{Addr: redisAddr})
    defer client.Close()

    // ------------------------------------------------------
    // 示例1:将任务入队以立即处理。
    //       使用(*Client).Enqueue方法。
    // ------------------------------------------------------

    task, err := tasks.NewEmailDeliveryTask(42, "some:template:id")
    if err != nil {
        log.Fatalf("无法创建任务: %v", err)
    }
    info, err := client.Enqueue(task)
    if err != nil {
        log.Fatalf("无法将任务入队: %v", err)
    }
    log.Printf("已入队任务: id=%s queue=%s", info.ID, info.Queue)


    // ------------------------------------------------------------
    // 示例2:安排任务在未来处理。
    //       使用ProcessIn或ProcessAt选项。
    // ------------------------------------------------------------

    info, err = client.Enqueue(task, asynq.ProcessIn(24*time.Hour))
    if err != nil {
        log.Fatalf("无法安排任务: %v", err)
    }
    log.Printf("已入队任务: id=%s queue=%s", info.ID, info.Queue)


    // ----------------------------------------------------------------------------
    // 示例3:设置其他选项以调整任务处理行为。
    //       选项包括MaxRetry、Queue、Timeout、Deadline、Unique等。
    // ----------------------------------------------------------------------------

    task, err = tasks.NewImageResizeTask("https://example.com/myassets/image.jpg")
    if err != nil {
        log.Fatalf("无法创建任务: %v", err)
    }
    info, err = client.Enqueue(task, asynq.MaxRetry(10), asynq.Timeout(3 * time.Minute))
    if err != nil {
        log.Fatalf("无法将任务入队: %v", err)
    }
    log.Printf("已入队任务: id=%s queue=%s", info.ID, info.Queue)
}

package main

import (
    "log"

    "github.com/hibiken/asynq"
    "your/app/package/tasks"
)

const redisAddr = "127.0.0.1:6379"

func main() {
    srv := asynq.NewServer(
        asynq.RedisClientOpt{Addr: redisAddr},
        asynq.Config{
            // 指定要使用的并发工作者数量
            Concurrency: 10,
            // 可选择指定具有不同优先级的多个队列。
            Queues: map[string]int{
                "critical": 6,
                "default":  3,
                "low":      1,
            },
            // 有关其他配置选项,请参阅godoc
        },
    )

    // mux将类型映射到处理程序
    mux := asynq.NewServeMux()
    mux.HandleFunc(tasks.TypeEmailDelivery, tasks.HandleEmailDeliveryTask)
    mux.Handle(tasks.TypeImageResize, tasks.NewImageProcessor())
    // ...注册其他处理程序...

    if err := srv.Run(mux); err != nil {
        log.Fatalf("无法运行服务器: %v", err)
    }
}