Golang 中的可靠后台任务处理：分布式消息队列实践

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"

    "github.com/go-redis/redis/v8" // 引入go-redis客户端库
)

// Task 定义一个示例任务结构
type EmailTask struct {
    Recipient string `json:"recipient"`
    Subject   string `json:"subject"`
    Body      string `json:"body"`
}

// NewRedisClient 创建并返回一个Redis客户端
func NewRedisClient() *redis.Client {
    rdb := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379", // Redis服务器地址
        Password: "",               // Redis密码，如果没有则为空
        DB:       0,                // DB号
    })
    return rdb
}

// PublishTask 将任务发布到Redis队列
func PublishTask(ctx context.Context, rdb *redis.Client, queueName string, task EmailTask) error {
    taskBytes, err := json.Marshal(task)
    if err != nil {
        return fmt.Errorf("failed to marshal task: %w", err)
    }

    // LPUSH 将任务推送到列表的左侧（头部）
    // 在简单的队列场景中，通常使用LPUSH/RPUSH作为生产者，BRPOP/BLPOP作为消费者
    err = rdb.LPush(ctx, queueName, taskBytes).Err()
    if err != nil {
        return fmt.Errorf("failed to publish task to Redis: %w", err)
    }
    log.Printf("Task published to queue '%s': %+v", queueName, task)
    return nil
}

func main() {
    ctx := context.Background()
    rdb := NewRedisClient()

    // 模拟用户注册后发送确认邮件的任务
    task1 := EmailTask{Recipient: "user1@example.com", Subject: "Welcome!", Body: "Thank you for registering."}
    task2 := EmailTask{Recipient: "user2@example.com", Subject: "Action Required", Body: "Please confirm your account."}

    queueName := "email_queue"

    if err := PublishTask(ctx, rdb, queueName, task1); err != nil {
        log.Fatalf("Error publishing task1: %v", err)
    }
    if err := PublishTask(ctx, rdb, queueName, task2); err != nil {
        log.Fatalf("Error publishing task2: %v", err)
    }

    fmt.Println("Producer finished publishing tasks.")
}

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "time"

    "github.com/go-redis/redis/v8"
)

// Task 定义与生产者相同的任务结构
type EmailTask struct {
    Recipient string `json:"recipient"`
    Subject   string `json:"subject"`
    Body      string `json:"body"`
}

// NewRedisClient 创建并返回一个Redis客户端
func NewRedisClient() *redis.Client {
    rdb := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "",
        DB:       0,
    })
    return rdb
}

// ProcessTask 模拟处理任务的函数
func ProcessTask(task EmailTask) error {
    log.Printf("Processing email for %s: Subject='%s'", task.Recipient, task.Subject)
    // 模拟耗时操作，例如调用邮件服务API
    time.Sleep(2 * time.Second)
    // 模拟一定概率的失败
    if task.Recipient == "user2@example.com" {
        return fmt.Errorf("simulated error: failed to send email to %s", task.Recipient)
    }
    log.Printf("Successfully sent email to %s", task.Recipient)
    return nil
}

// ConsumeTasks 持续从Redis队列中消费任务
func ConsumeTasks(ctx context.Context, rdb *redis.Client, queueName string) {
    log.Printf("Worker started, listening on queue '%s'...", queueName)
    for {
        select {
        case <-ctx.Done():
            log.Println("Worker shutting down.")
            return
        default:
            // BRPOP 阻塞式地从列表的右侧（尾部）弹出元素
            // Timeout为0表示永远阻塞，直到有元素弹出
            result, err := rdb.BRPop(ctx, 0, queueName).Result()
            if err != nil {
                if err == redis.Nil { // 队列为空，BRPop会一直阻塞，不会返回redis.Nil
                    continue
                }
                log.Printf("Error consuming from Redis: %v", err)
                time.Sleep(time.Second) // 错误时稍作等待，避免CPU空转
                continue
            }

            // result[0]是队列名，result[1]是弹出的值
            taskBytes := []byte(result[1])
            var task EmailTask
            if err := json.Unmarshal(taskBytes, &task); err != nil {
                log.Printf("Error unmarshalling task: %v, raw data: %s", err, taskBytes)
                // 可以在此处将无法解析的消息放入死信队列
                continue
            }

            // 处理任务，并实现重试逻辑（此处简化，实际生产中应更完善）
            err = ProcessTask(task)
            if err != nil {
                log.Printf("Task processing failed for %+v: %v. Re-queueing...", task, err)
                // 任务处理失败，重新推回队列头部，以便稍后重试
                // 注意：简单的LPUSH可能导致死循环，生产环境应使用更复杂的重试策略，如延迟队列、重试次数限制等
                rdb.LPush(ctx, queueName, taskBytes)
            }
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    rdb := NewRedisClient()
    queueName := "email_queue"

    ConsumeTasks(ctx, rdb, queueName)
}