Go中的任务队列：实践指南

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

在现代开发中，经常需要异步处理任务，而不阻塞应用程序的主线程。任务队列（task queues）是一种经典模式，它允许将工作分配给多个处理器（worker），管理负载并确保容错性。

Go语言（Golang）提供了实现任务队列的内置机制：goroutine和channel。在本文中，我们将通过实践讲解如何使用Go标准库构建一个可靠的任务队列，而无需借助外部消息代理（尽管在生产环境中RabbitMQ或Redis可能很有用）。

主体部分

1. 基础Worker Pool模式

Worker Pool是一组固定数量的goroutine（worker），它们监听同一个channel并处理传入的任务。这样可以控制并行度，避免系统过载。

我们来看一个简单示例：创建一个包含3个worker的池，处理整数（模拟工作）。

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("Worker %d 开始任务 %d", id, job)        time.Sleep(time.Second) // 模拟工作        fmt.Printf("Worker %d 完成任务 %d", id, job)    }}

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

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

    // 启动worker    for w := 1; w <= numWorkers; w++ {        wg.Add(1)        go worker(w, jobs, &wg)    }

    // 发送任务    for j := 1; j <= numJobs; j++ {        jobs <- j    }    close(jobs) // 关闭channel，使worker退出

    wg.Wait()    fmt.Println("所有任务已完成")}

代码解析：

• jobs — 带缓冲的channel，用于发送任务。
• worker — 从channel读取并处理每个任务的函数。
• sync.WaitGroup — 等待所有worker完成。
• 发送完所有任务后，channel被关闭（close(jobs)），导致worker中的for range循环结束。

这种模式是更复杂解决方案的基础。

2. 带优先级和超时的队列

在实际应用中，任务可能有不同的优先级，或者需要在特定时间内完成。我们实现一个队列，其中高优先级任务先被处理，并添加带超时的上下文以取消耗时任务。

package main

import (    "context"    "fmt"    "time")

type Task struct {    ID       int    Priority int // 数字越小，优先级越高    Data     string}

func processTask(ctx context.Context, task Task) error {    select {    case <-ctx.Done():        return ctx.Err()    case <-time.After(time.Duration(task.Priority) * 500 * time.Millisecond):        fmt.Printf("已处理任务 %d: %s (优先级 %d)", task.ID, task.Data, task.Priority)        return nil    }}

func main() {    tasks := []Task{        {ID: 1, Priority: 3, Data: "低优先级"},        {ID: 2, Priority: 1, Data: "高优先级"},        {ID: 3, Priority: 2, Data: "中优先级"},    }

    // 按优先级排序（最高优先级在前）    // 在实际项目中，请使用排序或堆（container/heap）    for i := 0; i < len(tasks); i++ {        for j := i + 1; j < len(tasks); j++ {            if tasks[j].Priority < tasks[i].Priority {                tasks[i], tasks[j] = tasks[j], tasks[i]            }