如何在Golang中实现工作池模式_Golang 工作池模式实践-Golang-PHP中文网

如何在Golang中实现工作池模式_Golang 工作池模式实践

P粉602998670

发布： 2025-11-21 13:14:02

原创

729人浏览过

工作池模式通过限制goroutine数量高效处理并发任务。1. 使用有缓冲channel作为任务队列；2. 启动固定worker从队列消费任务；3. 主协程分发任务并用WaitGroup等待完成；4. 可扩展结果channel返回执行结果与错误。适用于批量I/O操作，需合理设置worker数、缓冲大小，及时关闭channel，避免泄漏。

如何在golang中实现工作池模式_golang 工作池模式实践

在Go语言中，工作池（Worker Pool）模式是一种常见的并发处理方式，适用于需要控制并发数量、高效处理大量短期任务的场景。通过限制启动的goroutine数量，既能充分利用系统资源，又能避免因创建过多协程导致内存溢出或调度开销过大。

工作池的基本结构

一个典型的工作池由以下几个部分组成：

任务队列：使用有缓冲的channel存放待处理的任务。
工作者（Worker）：固定数量的goroutine从任务队列中读取任务并执行。
任务分发机制：主协程将任务发送到任务队列。
等待所有任务完成：使用sync.WaitGroup确保所有任务执行完毕。

这种模式特别适合批量处理I/O操作，如HTTP请求、文件读写、数据库查询等。

实现一个简单的工作池

下面是一个可运行的示例，展示如何用Go实现一个基础工作池：

立即学习“go语言免费学习笔记（深入）”；

package main

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

// Task 表示一个待处理的任务
type Task struct {
    ID   int
    Data string
}

// Worker 处理任务的函数
func worker(id int, jobs <-chan Task, wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range jobs {
        fmt.Printf("Worker %d processing task %d: %s\n", id, task.ID, task.Data)
        time.Sleep(time.Second) // 模拟耗时操作
    }
}

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

    var wg sync.WaitGroup
    jobs := make(chan Task, numTasks)

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

    // 发送任务到任务队列
    for i := 1; i <= numTasks; i++ {
        jobs <- Task{ID: i, Data: fmt.Sprintf("data-%d", i)}
    }
    close(jobs) // 关闭channel，防止goroutine泄漏

    // 等待所有worker完成
    wg.Wait()
    fmt.Println("All tasks completed.")
}

登录后复制

在这个例子中，我们启动了3个worker，共提交10个任务。任务通过channel分发，每个worker循环接收任务直到channel关闭。

Tellers AI

Tellers是一款自动视频编辑工具，可以将文本、文章或故事转换为视频。

查看详情

增强版：支持结果返回和错误处理

实际应用中，任务往往需要返回结果或上报错误。可以通过增加结果channel来实现：

type Result struct {
    TaskID int
    Error  error
    Output string
}

func workerWithResult(id int, jobs <-chan Task, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for task := range jobs {
        var err error
        output := fmt.Sprintf("processed-%s", task.Data)

        // 模拟可能失败的操作
        if task.ID%5 == 0 {
            err = fmt.Errorf("task %d failed intentionally", task.ID)
        }

        results <- Result{
            TaskID: task.ID,
            Error:  err,
            Output: output,
        }
    }
}

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

    var wg sync.WaitGroup
    jobs := make(chan Task, numTasks)
    results := make(chan Result, numTasks)

    // 启动带结果返回的worker
    for i := 1; i <= numWorkers; i++ {
        wg.Add(1)
        go workerWithResult(i, jobs, results, &wg)
    }

    // 提交任务
    for i := 1; i <= numTasks; i++ {
        jobs <- Task{ID: i, Data: fmt.Sprintf("data-%d", i)}
    }
    close(jobs)

    // 单独启动一个goroutine收集结果
    go func() {
        wg.Wait()
        close(results)
    }()

    // 打印结果
    for result := range results {
        if result.Error != nil {
            fmt.Printf("Task %d failed: %v\n", result.TaskID, result.Error)
        } else {
            fmt.Printf("Task %d succeeded: %s\n", result.TaskID, result.Output)
        }
    }

    fmt.Println("All tasks with results completed.")
}

登录后复制

这个版本增加了结果回传机制，主协程可以统一处理成功或失败的任务，便于日志记录、重试或通知。