第84课 - 实战：分布式任务队列

学习目标

通过本课学习，你将能够：

理解分布式任务队列的架构与价值：掌握为什么需要分布式任务队列以及其解决的核心问题。
掌握使用Redis作为任务队列：学会利用Redis的List结构和BRPOP命令实现一个稳定、高性能的任务队列。
实现任务的生产者和消费者：编写能够异步产生任务和并发处理任务的Go代码。
构建健壮的Worker池：使用Goroutine和Channel控制并发度，并实现优雅的启动与关闭。
实践错误处理与重试机制：为任务执行添加基础的错误处理逻辑。

核心概念

在上一课的WebSocket聊天室中，我们处理的是实时、双向的通信。但在很多业务场景下，例如发送邮件、生成报告、处理图片等，这些任务耗时较长，不需要立即返回结果。如果让用户等待，体验会非常差。

分布式任务队列应运而生。它的思想很简单：主程序（生产者）将“待办事项”放入一个共享的、中心化的“任务列表”（队列）中，然后立即返回。一组独立的程序（消费者/Worker）负责从这个列表中取出任务并执行。

为什么选择Redis？ Redis的List数据结构天生就是一个双端队列，其LPUSH和BRPOP命令组合，可以完美实现一个先进先出（FIFO）的任务队列。BRPOP是阻塞式的，这意味着当队列为空时，Worker不会空转（忙等），而是被阻塞，这能有效节省CPU资源，直到有新任务到来。

简单来说，架构如下：

[生产者应用] -- LPUSH --> [Redis List (队列)] -- BRPOP --> [Worker 1]
                           [Redis List (队列)] -- BRPOP --> [Worker 2]
                           ...                             ...

代码示例

我们将构建一个模拟的分布式邮件发送任务队列。生产者负责将“发送邮件”的任务信息序列化后推入队列，多个Worker并发地从队列中取出任务并执行（这里用打印模拟）。

项目结构：

project/
├── go.mod
├── main.go
├── producer/
│   └── producer.go
├── consumer/
│   └── consumer.go
└── utils/
    └── task.go

1. 安装Redis依赖：

go get github.com/go-redis/redis/v8

2. 定义任务结构 (utils/task.go)：

package utils

import "time"

// Task 代表一个待处理的任务
type Task struct {
    ID        string    `json:"id"`
    Payload   string    `json:"payload"` // 实际的任务数据，如邮件内容
    CreatedAt time.Time `json:"created_at"`
}

// NewTask 创建一个新任务
func NewTask(id, payload string) *Task {
    return &Task{
        ID:        id,
        Payload:   payload,
        CreatedAt: time.Now(),
    }
}

3. 任务生产者 (producer/producer.go)：

package producer

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "project/utils"
    "github.com/go-redis/redis/v8"
)

const TaskQueueName = "email_tasks"

// Producer 封装了生产任务的能力
type Producer struct {
    client *redis.Client
}

func NewProducer(client *redis.Client) *Producer {
    return &Producer{client: client}
}

// Enqueue 将任务推入Redis队列
func (p *Producer) Enqueue(ctx context.Context, task *utils.Task) error {
    // 将任务结构体序列化为JSON字节
    data, err := json.Marshal(task)
    if err != nil {
        return fmt.Errorf("failed to marshal task: %w", err)
    }

    // 使用LPUSH将任务推入队列列表的头部
    err = p.client.LPush(ctx, TaskQueueName, data).Err()
    if err != nil {
        return fmt.Errorf("failed to push task to queue: %w", err)
    }

    log.Printf("[Producer] Enqueued task %s", task.ID)
    return nil
}

4. 任务消费者/Worker (consumer/consumer.go)：

package consumer

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "project/utils"
    "time"
    "github.com/go-redis/redis/v8"
)

// Worker 代表一个消费者Worker
type Worker struct {
    id     int
    client *redis.Client
}

func NewWorker(id int, client *redis.Client) *Worker {
    return &Worker{id: id, client: client}
}

// Start 启动Worker，开始监听队列
func (w *Worker) Start(ctx context.Context) {
    log.Printf("[Worker %d] Started, waiting for tasks...", w.id)

    for {
        select {
        case <-ctx.Done():
            log.Printf("[Worker %d] Shutting down...", w.id)
            return
        default:
            // 关键：使用BRPOP从队列尾部阻塞式地弹出一个任务，超时时间为0（无限等待）
            result, err := w.client.BRPop(ctx, 0*time.Second, TaskQueueName).Result()
            if err != nil {
                if err == context.Canceled || err == context.DeadlineExceeded {
                    log.Printf("[Worker %d] Context canceled, exiting.", w.id)
                    return
                }
                log.Printf("[Worker %d] Error popping task: %v", w.id, err)
                continue
            }
            // BRPOP返回的结果是一个字符串切片 [key, value]
            taskData := result[1]
            w.processTask(taskData)
        }
    }
}

// processTask 处理单个任务
func (w *Worker) processTask(data string) {
    var task utils.Task
    if err := json.Unmarshal([]byte(data), &task); err != nil {
        log.Printf("[Worker %d] Failed to unmarshal task: %v", w.id, err)
        return // 无法解析，任务作废
    }

    log.Printf("[Worker %d] Processing task %s: %s", w.id, task.ID, task.Payload)

    // 模拟耗时任务（如发送邮件）
    time.Sleep(2 * time.Second)

    log.Printf("[Worker %d] Finished task %s", w.id, task.ID)
}

5. 主程序 (main.go)：

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "project/consumer"
    "project/producer"
    "project/utils"
    "sync"
    "syscall"
    "time"
    "github.com/go-redis/redis/v8"
)

func main() {
    // 初始化Redis客户端
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
        DB:   0,
    })
    defer rdb.Close()

    // 创建一个可取消的Context，用于优雅关闭
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // 启动多个Worker（消费者）
    var wg sync.WaitGroup
    numWorkers := 3
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            worker := consumer.NewWorker(id, rdb)
            worker.Start(ctx)
        }(i)
    }

    // 启动一个生产者，模拟产生任务
    prod := producer.NewProducer(rdb)
    go func() {
        for i := 1; i <= 10; i++ {
            task := utils.NewTask(
                fmt.Sprintf("task-%d", i),
                fmt.Sprintf("Send welcome email to user_%[email protected]", i),
            )
            if err := prod.Enqueue(ctx, task); err != nil {
                log.Printf("Error enqueueing task: %v", err)
            }
            time.Sleep(500 * time.Millisecond) // 模拟任务产生间隔
        }
    }()

    // 等待中断信号以优雅关闭
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
    <-sigChan
    log.Println("Received shutdown signal, canceling context...")

    // 发送取消信号，通知所有Worker退出
    cancel()

    // 等待所有Worker完成当前任务并退出
    wg.Wait()
    log.Println("All workers stopped. Exiting.")
}

运行说明：

确保本地Redis服务运行。
运行 go run main.go。
你将看到3个Worker启动，然后生产者开始产生任务，Worker们会轮流消费任务。
按下 Ctrl+C，程序会优雅地关闭所有Worker。

实践练习

练习1：增加任务优先级 修改代码，支持两个队列：high_priority_tasks 和 low_priority_tasks。修改Worker，使其优先从高优先级队列获取任务（提示：在BRPOP命令中传入多个键，它会按顺序检查）。

练习2：实现简单的重试机制 如果Worker在处理任务时发生错误（模拟：随机返回错误），将该任务重新推回队列，并为其添加一个retry_count字段。当retry_count超过3次时，将任务放入一个dead_letter_queue（死信队列）而不是重试。

练习3：添加任务执行监控 使用Redis的另一个数据结构（如SET）来记录正在被处理的任务（任务ID）。任务处理开始时添加，结束或失败时移除。你可以写一个简单的函数来查询当前正在处理的任务列表。

常见错误

忘记处理Redis连接错误：生产环境网络不稳定，必须妥善处理连接断开、超时等问题。
无限制的Worker并发：示例中启动了3个固定Worker。在实际应用中，应使用带缓冲的Channel或专门的Worker池库（如ants）来管理并发数，避免资源耗尽。
未序列化/反序列化复杂数据：任务Payload如果是结构体，一定要正确地使用json.Marshal/Unmarshal。
阻塞操作未使用Context：BRPOP的阻塞可能无限期持续，必须传入一个context.Context，以便在需要时（如程序退出）能够中断它。
忽略任务执行结果：本例只模拟了“即发即忘”的任务。如果任务需要结果（如生成报告后返回文件路径），则需要另一个机制（如回调、另一个结果队列）来传递。

小结

本课我们通过一个完整的项目，实现了基于Redis的Go分布式任务队列：

核心模式：生产者-消费者模式，解耦任务产生与任务执行。
Redis角色：利用List和BRPOP实现了一个可靠、高性能、阻塞式的消息队列。
并发控制：通过启动多个Goroutine作为Worker，并使用context实现优雅启停。
实践价值：这种模式极大地提升了Web应用的响应速度和系统的可扩展性。耗时任务被异步处理，Web请求可以快速返回；当任务量增加时，只需水平扩展Worker数量即可。

在实际项目中，你还需要考虑任务持久化（Redis AOF/RDB）、监控队列深度、Worker健康检查、更完善的错误处理与重试策略，以及可能使用更专业的消息队列中间件（如RabbitMQ, Kafka）来满足更复杂的需求。

实战：分布式任务队列