使用Asynq构建Golang分布式任务队列

Asynq是一个基于Redis的Go语言分布式任务队列库，它简单、可靠且高效。下面我将详细介绍Asynq的核心概念和使用方法。

Asynq核心概念

1. Task：表示要执行的工作单元
2. Client：用于创建和调度任务
3. Server：处理任务的工作进程
4. Worker：实际执行任务的处理器
5. Queue：任务存储的队列（默认队列名为"default"）

安装Asynq

go get -u github.com/hibiken/asynq

基本使用示例

1. 创建任务

package main

import (
	"fmt"
	"log"

	"github.com/hibiken/asynq"
)

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

// EmailDeliveryPayload 定义邮件任务负载
type EmailDeliveryPayload struct {
	UserID     int
	TemplateID string
}

// ImageResizePayload 定义图片处理任务负载
type ImageResizePayload struct {
	SourceURL string
	Width     int
	Height    int
}

func main() {
	// 创建Redis连接选项
	redisConnOpt := asynq.RedisClientOpt{
		Addr: "localhost:6379", // Redis地址
	}

	// 创建Asynq客户端
	client := asynq.NewClient(redisConnOpt)
	defer client.Close()

	// 创建邮件任务
	emailTask, err := asynq.NewTask(
		TypeEmailDelivery,
		EmailDeliveryPayload{UserID: 42, TemplateID: "welcome"},
	)
	if err != nil {
		log.Fatalf("could not create task: %v", err)
	}

	// 将任务加入队列
	info, err := client.Enqueue(emailTask)
	if err != nil {
		log.Fatalf("could not enqueue task: %v", err)
	}
	fmt.Printf("enqueued task: id=%s queue=%s\n", info.ID, info.Queue)

	// 创建图片处理任务（带选项）
	imgTask, err := asynq.NewTask(
		TypeImageResize,
		ImageResizePayload{SourceURL: "http://example.com/image.jpg", Width: 300, Height: 200},
	)
	if err != nil {
		log.Fatalf("could not create task: %v", err)
	}

	// 带选项的任务入队
	info, err = client.Enqueue(
		imgTask,
		asynq.Queue("critical"), // 指定队列
		asynq.MaxRetry(5),       // 最大重试次数
		asynq.Timeout(20*time.Minute), // 超时时间
	)
	if err != nil {
		log.Fatalf("could not enqueue task: %v", err)
	}
	fmt.Printf("enqueued task: id=%s queue=%s\n", info.ID, info.Queue)
}

2. 创建任务处理器

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/hibiken/asynq"
)

// NewEmailTaskProcessor 创建邮件任务处理器
func NewEmailTaskProcessor() asynq.Handler {
	return asynq.HandlerFunc(func(ctx context.Context, task *asynq.Task) error {
		var p EmailDeliveryPayload
		if err := asynq.UnmarshalPayload(task.Payload(), &p); err != nil {
			return fmt.Errorf("unmarshal payload failed: %v", err)
		}
		log.Printf("sending email to user %d using template %s", p.UserID, p.TemplateID)
		// 这里实现实际的邮件发送逻辑
		return nil
	})
}

// NewImageTaskProcessor 创建图片处理任务处理器
func NewImageTaskProcessor() asynq.Handler {
	return asynq.HandlerFunc(func(ctx context.Context, task *asynq.Task) error {
		var p ImageResizePayload
		if err := asynq.UnmarshalPayload(task.Payload(), &p); err != nil {
			return fmt.Errorf("unmarshal payload failed: %v", err)
		}
		log.Printf("resizing image from %s to %dx%d", p.SourceURL, p.Width, p.Height)
		// 这里实现实际的图片处理逻辑
		return nil
	})
}

func main() {
	// 创建Redis连接选项
	redisConnOpt := asynq.RedisClientOpt{
		Addr: "localhost:6379",
	}

	// 创建Asynq服务器
	srv := asynq.NewServer(
		redisConnOpt,
		asynq.Config{
			Concurrency: 10, // 最大并发数
			Queues: map[string]int{
				"critical": 6, // 关键队列权重
				"default":  3,  // 默认队列权重
				"low":      1,  // 低优先级队列权重
			},
		},
	)

	// 创建任务多路复用器
	mux := asynq.NewServeMux()
	mux.HandleFunc(TypeEmailDelivery, NewEmailTaskProcessor())
	mux.HandleFunc(TypeImageResize, NewImageTaskProcessor())

	// 启动服务器
	if err := srv.Run(mux); err != nil {
		log.Fatalf("could not run server: %v", err)
	}
}

高级功能

1. 定时任务

// 在未来某个时间执行任务
info, err := client.Enqueue(
    emailTask,
    asynq.ProcessIn(24*time.Hour), // 24小时后执行
)

// 定期任务（需要Asynq Monitor服务）
// 首先创建调度器
scheduler := asynq.NewScheduler(redisConnOpt, &asynq.SchedulerOpts{})

// 注册定期任务
entryID, err := scheduler.Register(
    "@every 1h", // 每小时执行一次
    emailTask,
)
if err != nil {
    log.Fatalf("could not schedule task: %v", err)
}

// 启动调度器
if err := scheduler.Run(); err != nil {
    log.Fatalf("could not run scheduler: %v", err)
}

2. 任务重试机制

Asynq内置了自动重试机制，可以通过选项配置：

info, err := client.Enqueue(
    task,
    asynq.MaxRetry(3),       // 最大重试次数
    asynq.RetryDelay(2*time.Minute), // 重试间隔
    asynq.Timeout(10*time.Minute), // 任务超时时间
)

3. 任务优先级

通过设置不同队列的权重来实现优先级：

srv := asynq.NewServer(
    redisConnOpt,
    asynq.Config{
        Concurrency: 10,
        Queues: map[string]int{
            "critical": 6, // 60%的资源
            "default":  3,  // 30%的资源
            "low":      1,  // 10%的资源
        },
    },
)

监控和管理

Asynq提供了Web UI和CLI工具来监控和管理任务：

1. Web UI：

http.Handle("/", asynqmon.New(asynqmon.Options{
    RootPath:     "/monitoring", // 根路径
    RedisConnOpt: redisConnOpt,
}))
log.Fatal(http.ListenAndServe(":8080", nil))

1. CLI工具：

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

# 查看队列统计
asynq stats --redis-addr=localhost:6379

# 查看待处理任务
asynq queue list --redis-addr=localhost:6379

最佳实践

1. 任务设计：

   • 保持任务小而专注
   • 任务应该是幂等的
   • 合理设置任务超时时间

2. 错误处理：

   • 在处理器中妥善处理错误
   • 记录失败任务以便排查
   • 设置合理的重试策略

3. 性能优化：

   • 根据服务器资源调整并发数
   • 对任务进行合理分类和优先级划分
   • 监控队列积压情况

Asynq是一个功能强大但易于使用的分布式任务队列库，非常适合需要可靠后台任务处理的Go应用程序。通过合理配置，它可以处理从简单到复杂的各种任务处理需求。