Asynq：Go 语言分布式任务队列，基于 Redis 构建的生产级方案

异步任务处理是后端开发的刚需——发邮件、生成报表、图片处理、推送通知，这些耗时操作不能阻塞主请求。Go 生态里，Asynq 是最成熟的分布式任务队列方案。

Asynq 是什么

Asynq 是一个用 Go 编写的分布式任务队列库，基于 Redis 构建。设计理念：

简单可靠：API 简洁，上手快
高效可扩展：支持水平扩展，适合高并发场景
生产可用：监控、重试、定时任务等完整功能

项目地址：https://github.com/hibiken/asynq

核心特性

可靠的任务执行

至少执行一次（At-least-once）语义，确保任务不会丢失。即使 Worker 崩溃，任务也会自动恢复。

灵活的调度能力

支持立即执行、延迟执行、定时执行三种模式：

// 立即执行
client.Enqueue(task)

// 24小时后执行
client.Enqueue(task, asynq.ProcessIn(24*time.Hour))

// 定时任务
scheduler.Register("@daily", task)

智能重试机制

任务失败自动重试，支持自定义重试次数和策略：

task := asynq.NewTask("email:send", payload,
    asynq.MaxRetry(5),
    asynq.Timeout(3*time.Minute),
)

优先级队列

支持多个队列，每个队列可以设置不同权重：

Queues: map[string]int{
    "critical": 6,  // 最高优先级
    "default":  3,
    "low":      1,
}

快速上手：5 分钟跑通

第一步：安装

go get -u github.com/hibiken/asynq

需要 Redis 4.0 或更高版本。

第二步：定义任务

package tasks

const TypeEmailDelivery = "email:deliver"

func NewEmailTask(userID int, templateID string) (*asynq.Task, error) {
    payload := map[string]interface{}{
        "user_id":     userID,
        "template_id": templateID,
    }
    data, _ := json.Marshal(payload)
    return asynq.NewTask(TypeEmailDelivery, data), nil
}

第三步：入队任务

client := asynq.NewClient(asynq.RedisClientOpt{Addr: "127.0.0.1:6379"})
defer client.Close()

task, _ := tasks.NewEmailTask(42, "welcome")
client.Enqueue(task)

第四步：处理任务

srv := asynq.NewServer(
    asynq.RedisClientOpt{Addr: "127.0.0.1:6379"},
    asynq.Config{Concurrency: 10},
)

mux := asynq.NewServeMux()
mux.HandleFunc(tasks.TypeEmailDelivery, handleEmailTask)

srv.Run(mux)

就是这么简单。

生产级监控

Asynqmon（Web UI）

功能完善的 Web 管理界面：

查看队列状态
搜索和查看任务详情
手动重试或删除任务
暂停/恢复队列
Prometheus 监控指标

命令行工具

# 安装
go install github.com/hibiken/asynq/tools/asynq@latest

# 查看仪表盘
asynq dash

# 查看队列状态
asynq queue ls

与其他方案对比

特性	Asynq	Celery (Python)	Sidekiq (Ruby)	Kafka	RabbitMQ
语言	Go	Python	Ruby	多语言	多语言
后端	Redis	Redis/RabbitMQ	Redis	自有	自有
定位	任务队列	任务队列	任务队列	消息流	消息队列
复杂度	低	中	低	高	高
运维成本	低	中	低	高	高
适用场景	Go 项目异步任务	Python 生态	Ruby 生态	数据流处理	企业级消息

实践建议

合理设置并发数

asynq.Config{
    Concurrency: runtime.NumCPU() * 10,  // 根据实际情况调整
}

善用中间件

func LoggingMiddleware(h asynq.Handler) asynq.Handler {
    return asynq.HandlerFunc(func(ctx context.Context, t *asynq.Task) error {
        start := time.Now()
        err := h.ProcessTask(ctx, t)
        log.Printf("Task %s took %v", t.Type(), time.Since(start))
        return err
    })
}