<img src="https://yellow-cdn.veclightyear.com/835a84d5/42647fa5-9655-4850-89db-eadba694f0c9.png" alt="Asynq logo" width="360px" />

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

Asynq是一个Go语言库，用于队列任务并通过工作者异步处理。它基于Redis，设计上既可扩展又易于上手。

Asynq的工作原理概述：

• 客户端将任务放入队列
• 服务器从队列中提取任务，并为每个任务启动一个工作者goroutine
• 多个工作者并发处理任务

任务队列用作在多台机器间分配工作的机制。系统可以包含多个工作者服务器和代理，从而实现高可用性和水平扩展。

用例示例

特性

• 保证任务至少执行一次
• 任务调度
• 失败任务的重试
• 工作者崩溃时自动恢复任务
• 加权优先队列
• 严格优先队列
• 由于Redis写入速度快，添加任务的延迟低
• 使用唯一选项实现任务去重
• 允许每个任务设置超时和截止时间
• 允许聚合任务组以批处理多个连续操作
• 灵活的处理程序接口，支持中间件
• 能够暂停队列以停止处理队列中的任务
• 周期性任务
• 支持Redis集群实现自动分片和高可用性
• 支持Redis哨兵实现高可用性
• 与Prometheus集成以收集和可视化队列指标
• Web UI用于检查和远程控制队列和任务
• 命令行工具用于检查和远程控制队列和任务

稳定性和兼容性

状态：该库目前正在大力开发中，API经常发生重大变化。

☝️ 重要提示：当前主版本为零（v0.x.x），以适应快速开发和快速迭代，同时获取用户的早期反馈（欢迎对API提供反馈！）。在v1.0.0发布之前，公共API可能会在没有主要版本更新的情况下发生变化。

赞助

如果您在生产环境中使用此软件包，请考虑赞助该项目以表示您的支持！

快速入门

确保您已安装Go（下载）。支持最新的两个Go版本（参见https://go.dev/dl）。

通过创建一个文件夹并在文件夹内运行go mod init github.com/your/repo（了解更多）来初始化您的项目。然后使用go get命令安装Asynq库：

go get -u github.com/hibiken/asynq

确保您在本地或从Docker容器中运行Redis服务器。需要4.0或更高版本。

接下来，编写一个封装任务创建和任务处理的包。

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失败: %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失败: %v: %w", err, asynq.SkipRetry)
    }
    log.Printf("正在调整图像大小: src=%s", p.SourceURL)
    // 图像调整大小代码 ...
    return nil
}

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

在您的应用程序代码中，导入上述包并使用Client将任务放入队列。

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") 如果err不为nil { log.Fatalf("无法创建任务: %v", err) } info, err = client.Enqueue(task, asynq.MaxRetry(10), asynq.Timeout(3 * time.Minute)) 如果err不为nil { log.Fatalf("无法将任务加入队列: %v", err) } log.Printf("已将任务加入队列: id=%s queue=%s", info.ID, info.Queue) }

接下来，启动一个工作服务器在后台处理这些任务。要启动后台工作者，请使用[`Server`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Server)并提供您的[`Handler`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#Handler)来处理任务。

您可以选择使用[`ServeMux`](https://pkg.go.dev/github.com/hibiken/asynq?tab=doc#ServeMux)创建处理程序，就像使用[`net/http`](https://golang.org/pkg/net/http/)处理程序一样。

```go
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)
    }
}

有关该库更详细的演练，请参阅我们的入门指南。

要了解更多关于asynq功能和API的信息，请参阅包的godoc。

Web界面

Asynqmon是一个基于Web的工具，用于监控和管理Asynq队列和任务。

以下是Web界面的几个截图：

队列视图

任务视图

指标视图 <img width="1532" alt="屏幕截图 2021-12-19 下午4:37:19" src="https://yellow-cdn.veclightyear.com/835a84d5/25478a6f-420f-4484-9d64-e40c7bb63a95.png">

设置和自适应暗模式

有关如何使用该工具的详细信息，请参阅工具的README。

命令行工具

Asynq附带一个命令行工具，用于检查队列和任务的状态。

要安装CLI工具，请运行以下命令：

go install github.com/hibiken/asynq/tools/asynq@latest

以下是运行asynq dash命令的示例：

有关如何使用该工具的详细信息，请参阅工具的README。

贡献

我们对社区做出的任何贡献（GitHub问题/PR、Gitter频道上的反馈等）都持开放态度并表示感谢。

在贡献之前，请参阅贡献指南。

许可证

版权所有 (c) 2019-现在 Ken Hibino和贡献者。Asynq是根据MIT许可证授权的免费开源软件。官方logo由Vic Shóstak创建，并根据知识共享许可证（CC0 1.0通用）分发。