Go语言中的可靠后台任务处理：分布式队列实践

package main

import (
    "fmt"
    "time"
)

func sendConfirmationEmail(userEmail string) {
    fmt.Printf("模拟发送邮件到: %s...\n", userEmail)
    time.Sleep(5 * time.Second) // 模拟邮件发送耗时
    fmt.Printf("邮件发送完成给: %s\n", userEmail)
}

func main() {
    userEmail := "test@example.com"
    go sendConfirmationEmail(userEmail) // 在goroutine中发送邮件

    fmt.Println("用户注册成功，主程序继续执行...")
    // 主程序可能在邮件发送完成前退出
    time.Sleep(6 * time.Second) // 确保有足够时间观察goroutine输出
}

package main

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

    // 假设这里引入了某个队列客户端库，例如 for RabbitMQ, Beanstalkd, or Redis
    // import "github.com/streadway/amqp" (for RabbitMQ)
    // import "github.com/beanstalkd/go-beanstalk" (for Beanstalkd)
    // import "github.com/go-redis/redis/v8" (for Redis)
)

// EmailJob 定义了邮件发送任务的数据结构
type EmailJob struct {
    Recipient string `json:"recipient"`
    Subject   string `json:"subject"`
    Body      string `json:"body"`
}

// 模拟一个队列客户端接口
type QueueClient interface {
    Enqueue(jobType string, payload []byte) error
    Dequeue(jobType string) ([]byte, error)
    Acknowledge(jobID string) error // 任务完成确认
    // ... 其他方法如重试、死信队列等
}

// 模拟具体的队列客户端实现 (这里以一个简单的内存队列为例，实际应替换为真实的分布式队列客户端)
type InMemoryQueue struct {
    queue chan []byte
}

func NewInMemoryQueue() *InMemoryQueue {
    return &InMemoryQueue{
        queue: make(chan []byte, 100), // 缓冲区大小
    }
}

func (q *InMemoryQueue) Enqueue(jobType string, payload []byte) error {
    select {
    case q.queue <- payload:
        log.Printf("任务入队: %s", string(payload))
        return nil
    default:
        return fmt.Errorf("队列已满，无法入队")
    }
}

func (q *InMemoryQueue) Dequeue(jobType string) ([]byte, error) {
    select {
    case payload := <-q.queue:
        log.Printf("任务出队: %s", string(payload))
        return payload, nil
    case <-time.After(5 * time.Second): // 模拟阻塞等待
        return nil, fmt.Errorf("队列空，等待超时")
    }
}

func (q *InMemoryQueue) Acknowledge(jobID string) error {
    // 内存队列无需确认，真实队列需要
    return nil
}


// 生产者：将任务推送到队列
func produceEmailJob(qc QueueClient, recipient, subject, body string) error {
    job := EmailJob{
        Recipient: recipient,
        Subject:   subject,
        Body:      body,
    }
    payload, err := json.Marshal(job)
    if err != nil {
        return fmt.Errorf("序列化邮件任务失败: %w", err)
    }
    return qc.Enqueue("email_send", payload)
}

// 消费者：从队列中拉取任务并处理
func startWorker(qc QueueClient) {
    fmt.Println("邮件发送工作者启动...")
    for {
        payload, err := qc.Dequeue("email_send")
        if err != nil {
            log.Printf("从队列获取任务失败: %v", err)
            time.Sleep(1 * time.Second) // 短暂等待后重试
            continue
        }

        var job EmailJob
        if err := json.Unmarshal(payload, &job); err != nil {
            log.Printf("反序列化邮件任务失败: %v, 原始payload: %s", err, string(payload))
            // 记录错误，可能需要将此任务移至死信队列
            continue
        }

        // 执行实际的邮件发送逻辑
        fmt.Printf("工作者处理邮件任务 - 收件人: %s, 主题: %s\n", job.Recipient, job.Subject)
        time.Sleep(3 * time.Second) // 模拟实际发送耗时
        fmt.Printf("邮件发送成功给: %s\n", job.Recipient)

        // 确认任务完成，从队列中移除
        // 在真实队列中，这通常是调用队列客户端的ack方法
        _ = qc.Acknowledge("some-job-id-from-queue") // 假设队列会返回一个job ID
    }
}

func main() {
    // 初始化队列客户端 (实际应用中会连接到RabbitMQ, Beanstalkd, Redis等)
    queueClient := NewInMemoryQueue() // 替换为真实的队列客户端

    // 启动一个或多个消费者工作者
    go startWorker(queueClient)
    go startWorker(queueClient) // 可以启动多个工作者并发处理

    // 主程序作为生产者，生成任务
    fmt.Println("主程序开始生产邮件任务...")
    for i := 0; i < 5; i++ {
        recipient := fmt.Sprintf("user%d@example.com", i)
        subject := fmt.Sprintf("欢迎注册 %d", i)
        body := "感谢您的注册！"
        if err := produceEmailJob(queueClient, recipient, subject, body); err != nil {
            log.Printf("生产任务失败: %v", err)
        }
        time.Sleep(500 * time.Millisecond) // 模拟任务生产间隔
    }

    fmt.Println("主程序任务生产完成，等待工作者处理...")
    time.Sleep(20 * time.Second) // 确保工作者有足够时间处理任务
}