Golang任务队列与异步处理项目示例

type Task struct {
    ID      string
    Payload map[string]interface{}
    Process func(map[string]interface{}) error
}

const QueueSize = 100
const WorkerCount = 5
<p>var TaskQueue = make(chan Task, QueueSize)</p><p>func StartWorkers() {
for i := 0; i < WorkerCount; i++ {
go func(workerID int) {
for task := range TaskQueue {
println("Worker", workerID, "processing task:", task.ID)
err := task.Process(task.Payload)
if err != nil {
println("Task failed:", task.ID, "Error:", err.Error())
} else {
println("Task completed:", task.ID)
}
}
}(i)
}
}
</font></p><H3>2. 提交异步任务示例</H3><p>将具体业务逻辑封装为任务提交到队列，实现异步执行。</p><p>例如，模拟发送邮件任务：</p>
                    <div class="aritcle_card">
                        <a class="aritcle_card_img" href="/ai/2140">
                            <img src="https://img.php.cn/upload/ai_manual/000/000/000/175680088878918.png" alt="芦笋演示">
                        </a>
                        <div class="aritcle_card_info">
                            <a href="/ai/2140">芦笋演示</a>
                            <p>一键出成片的录屏演示软件，专为制作产品演示、教学课程和使用教程而设计。</p>
                            <div class="">
                                <img src="/static/images/card_xiazai.png" alt="芦笋演示">
                                <span>34</span>
                            </div>
                        </div>
                        <a href="/ai/2140" class="aritcle_card_btn">
                            <span>查看详情</span>
                            <img src="/static/images/cardxiayige-3.png" alt="芦笋演示">
                        </a>
                    </div>
                <font face="Courier New"><pre class="brush:php;toolbar:false;">
func SendEmailAsync(to, subject, body string) {
    task := Task{
        ID:      fmt.Sprintf("email-%d", time.Now().UnixNano()),
        Payload: map[string]interface{}{"to": to, "subject": subject, "body": body},
        Process: func(data map[string]interface{}) error {
            // 模拟耗时操作
            time.Sleep(2 * time.Second)
            fmt.Printf("? Sending email to %v: %v\n", data["to"], data["subject"])
            return nil
        },
    }
    TaskQueue <- task
}
</font><p>调用该函数不会阻塞主流程：</p><font face="Courier New"><pre class="brush:php;toolbar:false;">
SendEmailAsync("user@example.com", "Welcome!", "Thank you for signing up.")
println("Email task submitted, continuing...")

rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
<p>func PushTaskToRedis(task Task) error {
data, _ := json.Marshal(task)
return rdb.LPush("task_queue", data).Err()
}
</p>

func RedisWorker() {
    for {
        val, err := rdb.BLPop(0, "task_queue").Result()
        if err != nil && err != redis.Nil {
            log.Println("Redis error:", err)
            continue
        }
        if len(val) == 2 {
            var task Task
            json.Unmarshal([]byte(val[1]), &task)
            task.Process(task.Payload)
        }
    }
}

type Task struct {
    ID       string
    Payload  map[string]interface{}
    MaxRetry int
    Retried  int
    Process  func(map[string]interface{}) error
}
<p>// 在worker中判断重试次数
if err := task.Process(task.Payload); err != nil {
if task.Retried < task.MaxRetry {
task.Retried++
TaskQueue <- task // 重新入队
} else {
println("Task permanently f<a style="color:#f60; text-decoration:underline;" title="ai" href="https://www.php.cn/zt/17539.html" target="_blank">ai</a>led:", task.ID)
}
}
</p>