Go 任务队列：async、gocelery 和 machinery 库概览

Go (Golang) 任务队列：概览与实践

在现代 Web 应用和微服务中，经常需要异步执行资源密集型或长时间运行的操作：发送电子邮件、生成报告、处理图像、与外部 API 交互等。为此，需要使用任务队列（task queues）。在 Go (Golang) 生态系统中，有几个成熟的库可以实现这种架构。在本文中，我们将介绍三种最流行的工具：machinery、gocelery 和 asynq，分析它们的特点并提供代码示例。

为什么 Go 需要任务队列？

任务队列解决了几个关键问题：

• 职责分离： HTTP 处理程序无需等待后台任务完成。
• 可扩展性： 您可以在不同服务器上启动多个工作进程（worker）。
• 可靠性： 工作进程故障时，任务不会丢失，可以重新处理。
• 调度： 能够将任务执行延迟到特定时间。

1. Machinery — 强大且灵活的库

Machinery 是一个用 Go 从头编写的异步任务队列。它支持多种消息代理（Redis、RabbitMQ、AWS SQS）和结果存储（Redis、MongoDB、Memcache）。Machinery 提供了丰富的 API 用于创建任务、链（chains）、组（groups）和重复任务。

安装与示例

go get github.com/RichardKnox/machinery/v2

一个简单的工作进程和客户端示例：

package main

import (    "github.com/RichardKnox/machinery/v2"    "github.com/RichardKnox/machinery/v2/config"    "github.com/RichardKnox/machinery/v2/tasks")

func main() {    // 使用 Redis 代理的配置    cnf := &config.Config{        Broker:        "redis://localhost:6379",        DefaultQueue:  "machinery_tasks",        ResultBackend: "redis://localhost:6379",    }

    server, err := machinery.NewServer(cnf)    if err != nil {        panic(err)    }

    // 注册任务    err = server.RegisterTasks(map[string]interface{}{        "add": func(a, b int) (int, error) {            return a + b, nil        },    })    if err != nil {        panic(err)    }

    // 启动工作进程    worker := server.NewWorker("worker1", 2) // 2 个并发任务    go worker.Launch()

    // 发送任务    task := tasks.Signature{        Name: "add",        Args: []tasks.Arg{            {Type: "int", Value: 5},            {Type: "int", Value: 7},        },    }    result, _ := server.SendTask(&task)    // 等待结果    if res, err := result.Get(10); err == nil {        println("Result:", res.(int)) // 输出: 12    }}

优点： 支持复杂的任务图、重试机制、通过 Web 界面进行监控。

缺点： 配置相对复杂，文档有时落后于代码。

2. Asynq — 具有丰富 UI 的现代队列

Asynq 是一个基于 Redis 的库，由 Google 的开发人员开发。它提供了便捷的 API、内置调度、任务优先级以及用于监控的内置 Web 界面。Asynq 正在积极开发中，并拥有出色的文档。

安装与示例

go get github.com/hibiken/asynq

一个简单的客户端和工作进程示例：

package main

import (    "context"    "log"    "github.com/hibiken/asynq")

// 类型化任务const TypeEmailDelivery = "email:deliver"

func main() {    // 用于发送任务的客户端    client := asynq.NewClient(asynq.RedisClientOpt{Addr: "localhost:6379"})    defer client.Clos