Go 任务队列：创建、处理与优化

引言：为什么在Go中需要任务队列？

在现代后端开发中，经常会出现不需要立即响应用户的任务：发送电子邮件、生成报告、处理图片、与外部API同步。在HTTP请求中同步执行此类操作是一种不良实践，因为它会拖慢响应速度并增加服务器负载。

任务队列解决了这个问题。在Go（Golang）中，内置的并发机制——goroutine和通道——让您无需外部依赖即可构建可靠且高性能的队列。本文将探讨如何从头创建任务队列，通过工作池进行优化，并避免常见错误。

基础：Goroutine与通道

Go提供了两个用于处理队列的关键原语：

• Goroutine——使用go关键字启动的轻量级线程。
• 通道——用于在goroutine之间传输数据的类型化"管道"。

最简单的队列是一个从通道读取并处理任务的goroutine：

package main

import (    "fmt"    "time")

func worker(jobs <-chan int) {    for job := range jobs {        fmt.Printf("处理任务 %d", job)        time.Sleep(1 * time.Second) // 模拟工作    }}

func main() {    jobs := make(chan int, 10)

    // 启动工作协程    go worker(jobs)

    // 发送任务    for i := 1; i <= 5; i++ {        jobs <- i    }

    close(jobs) // 发送所有任务后关闭通道    time.Sleep(2 * time.Second)}

这段代码创建了一个按顺序处理任务的工作协程。但对于实际项目，需要可扩展性——即多个工作协程。

在Go中实现工作池

工作池是一种模式，我们启动固定数量的工作协程，每个协程从公共通道中获取任务。这使您能够控制并行度和系统负载。

基本实现

package main

import (    "fmt"    "sync"    "time")

func worker(id int, jobs <-chan int, wg *sync.WaitGroup) {    defer wg.Done()    for job := range jobs {        fmt.Printf("工作协程 %d: 任务 %d", id, job)        time.Sleep(500 * time.Millisecond)    }}

func main() {    const numWorkers = 3    const numJobs = 10

    jobs := make(chan int, numJobs)    var wg sync.WaitGroup

    // 启动工作协程    for i := 1; i <= numWorkers; i++ {        wg.Add(1)        go worker(i, jobs, &wg)    }

    // 发送任务    for j := 1; j <= numJobs; j++ {        jobs <- j    }    close(jobs)

    wg.Wait() // 等待所有工作协程完成    fmt.Println("所有任务完成")}

这里我们使用sync.WaitGroup来等待所有工作协程完成。带有numJobs缓冲区的通道允许在工作者尚未取走任务时无阻塞地发送所有任务。

动态添加任务

在实际应用中，任务是异步到达的——例如通过HTTP。以下是一个带有处理程序的示例：

package main

import (    "fmt"    "net/http"    "sync"    "time")

type Task struct {    ID   int    Data string}

var (    taskQueue = make(chan Task, 100)    wg        sync.WaitGroup)

func worker(id int) {    defer wg.Done()    for task := range taskQueue {        fmt.Printf("[工作协程 %d] 处理中: %s (ID=%d)", id, task.Data, task.ID)        time.Sleep(1 * time.Second)    }}

func handler(w http.ResponseWriter, r *http.Request) {    // 从请求中获取任务（已简化）    task := Task{ID: time.Now().Nanosecond(), Data: "新任务"}    taskQueue <- task    w.W