第 28 课：并发模式 - Worker Pool

1. 学习目标

完成本课后，你将能够：

理解 Worker Pool（工作池/线程池）模式的必要性和核心组成部分。
使用 Go 语言的 goroutine 和 channel 实现一个基础的、可配置的 Worker Pool。
掌握使用带缓冲的 channel 作为信号量（Semaphore）来精确控制并发度。
学会实现 Worker Pool 的优雅关闭（Graceful Shutdown），确保所有任务都被妥善处理。
能够识别和避免在实现 Worker Pool 时常见的错误。

2. 核心概念

为什么需要 Worker Pool？

在并发编程中，为每个任务都创建一个新的 goroutine 有时是不可取的。例如，如果你需要同时处理 10,000 个 HTTP 请求或计算任务，瞬间创建 10,000 个 goroutine 可能会导致：

资源耗尽：每个 goroutine 都需要一定的内存（栈空间），大量 goroutine 会消耗大量内存。
性能下降：过多的 goroutine 会增加 Go 调度器的开销，并可能导致 CPU 缓存未命中率升高。
资源限制：下游依赖（如数据库连接池、文件句柄、API 速率限制）通常无法承受如此高的并发访问。

Worker Pool 模式正是为了解决这个问题而生。它预先创建一组固定数量的 worker goroutine，这些 worker 从一个共享的任务队列中获取任务并执行。这样，我们就可以控制同时执行任务的并发数量。

组成部分

一个典型的 Worker Pool 由以下三部分构成：

任务队列（Task Queue）：一个 channel，用于存放待处理的任务。
Worker 池（Worker Pool）：一组固定的 goroutine，它们循环地从任务队列中读取任务并执行。
分发器（Dispatcher）：负责将外部产生的任务发送到任务队列中。

信号量（Semaphore）的作用

在 Worker Pool 中，Worker 的数量本身就是一种并发控制。但有时我们需要更精细的控制，例如限制同时进行 I/O 操作的 worker 数量。Go 中一个优雅的做法是使用带缓冲的 channel 作为信号量。

// 创建一个容量为 3 的信号量 channel，表示最多允许 3 个并发操作。
semaphore := make(chan struct{}, 3)

Worker 在开始执行真正的工作（尤其是阻塞式 I/O）前，先向 semaphore 发送一个值（semaphore <- struct{}{}），执行完毕后再接收一个值（<- semaphore）。这样，当信号量 channel 已满时，新的 worker 会阻塞，从而实现了并发控制。

优雅关闭

一个健壮的 Worker Pool 必须能够优雅地关闭，确保：

不再接受新任务。
已经进入任务队列的所有任务都被执行完毕。
所有 worker 正常退出。这通常通过 sync.WaitGroup 和关闭任务队列 channel 来实现。

3. 代码示例

下面是一个完整的 Worker Pool 实现，它模拟处理一系列耗时的任务（如处理图片、发送邮件等）。

package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

// Task 代表一个任务结构体，包含ID和需要执行的工作描述。
type Task struct {
	ID      int
	Payload string
}

// Result 代表任务执行的结果。
type Result struct {
	TaskID int
	Output string
	Err    error
}

// worker 是实际工作的函数。它从 jobs channel 接收任务，处理后将结果发送到 results channel。
// 它通过 `quit` channel 接收停止信号。
func worker(id int, jobs <-chan Task, results chan<- Result, quit <-chan struct{}, wg *sync.WaitGroup) {
	defer wg.Done() // Worker退出时，通知WaitGroup
	for {
		select {
		case job, ok := <-jobs:
			if !ok { // jobs channel 被关闭了，意味着没有更多任务
				fmt.Printf("[Worker %d] 任务通道已关闭，退出。\n", id)
				return
			}
			// 模拟执行任务的耗时（如I/O、计算）
			fmt.Printf("[Worker %d] 开始处理任务 %d: %s\n", id, job.ID, job.Payload)
			time.Sleep(time.Duration(rand.Intn(500)) * time.Millisecond) // 随机耗时
			// 生成结果并发送
			results <- Result{
				TaskID: job.ID,
				Output: fmt.Sprintf("任务 %d 由 Worker %d 处理完毕", job.ID, id),
			}
		case <-quit: // 收到退出信号
			fmt.Printf("[Worker %d] 收到退出信号，退出。\n", id)
			return
		}
	}
}

// dispatcher 负责分发任务到jobs通道，并管理Worker Pool的生命周期。
func dispatcher(tasks []Task, numWorkers int) {
	jobs := make(chan Task, len(tasks))     // 任务队列，缓冲区大小为任务总数
	results := make(chan Result, len(tasks)) // 结果队列
	quit := make(chan struct{})             // 退出信号通道

	var wg sync.WaitGroup

	// 1. 启动Worker池
	for i := 1; i <= numWorkers; i++ {
		wg.Add(1)
		go worker(i, jobs, results, quit, &wg)
	}

	// 2. 分发所有任务到jobs通道
	for _, task := range tasks {
		jobs <- task
	}
	// 关闭jobs通道，通知所有Worker：没有新任务了。
	close(jobs)
	fmt.Println("[Dispatcher] 所有任务已分发，任务通道已关闭。")

	// 3. 等待所有Worker完成（它们会在jobs通道关闭后自然退出）
	wg.Wait()
	fmt.Println("[Dispatcher] 所有Worker已退出。")

	// 4. 关闭结果通道和退出通道
	close(results)
	close(quit)

	// 5. 收集并打印结果
	fmt.Println("\n--- 所有结果 ---")
	for result := range results {
		fmt.Printf("结果: %s\n", result.Output)
	}
}

func main() {
	// 生成一些模拟任务
	numTasks := 10
	tasks := make([]Task, numTasks)
	for i := 0; i < numTasks; i++ {
		tasks[i] = Task{
			ID:      i + 1,
			Payload: fmt.Sprintf("数据包-%d", i+1),
		}
	}

	// 启动分发器，使用3个Worker
	fmt.Println("启动 Worker Pool，Worker数量: 3")
	dispatcher(tasks, 3)
	fmt.Println("主程序结束。")
}

代码说明：

worker 函数是核心，它是一个无限循环，通过 select 同时监听任务 jobs 和退出信号 quit。
dispatcher 函数是总控，它创建 channel，启动指定数量的 worker goroutine，分发任务，然后通过关闭 jobs channel 和等待 WaitGroup 来协调关闭。
当 jobs channel 被关闭后，所有空闲的 worker 会从 <-jobs 读取到 ok=false，从而优雅退出。

4. 实践练习

练习 1：基础调整

基于上面的代码，将 numWorkers 从 3 改为 1 和 5，分别运行并观察输出。描述 Worker 数量如何影响任务的执行顺序和总耗时（基于模拟的随机延迟）。

练习 2：错误处理与超时

修改 worker 函数，使其有 30% 的概率模拟任务失败（返回一个 error）。同时，在 dispatcher 中，为整个任务处理过程设置一个总超时（例如 2 秒）。如果在超时时间内任务未全部完成，则强制关闭程序。

要求：打印出成功和失败的任务结果。
提示：使用 time.After 和 select 来实现超时。

练习 3：带信号量的高级 Worker Pool

创建一个新的 Worker Pool 实现，但这次不限制 Worker 数量（理论上每个任务一个 goroutine），而是使用一个容量为 3 的信号量 channel 来限制同时进行耗时 I/O 操作的 goroutine 数量。

要求：在 worker 开始真正的“工作”（time.Sleep）前获取信号量，工作完成后释放信号量。
预期行为：应该会有超过 3 个 goroutine 被创建，但同时进入 Sleep 状态的不会超过 3 个。

5. 常见错误

WaitGroup 计数器不匹配：忘记在每个 worker 启动时调用 wg.Add(1)，或忘记在 worker 退出前调用 wg.Done()，导致 wg.Wait() 永久阻塞。
信号量 channel 初始化错误：将信号量 channel 创建为无缓冲 channel (make(chan struct{}))，这会导致信号量完全失效，因为它一次只能通过一个值，失去了“池”的意义。
未关闭 channel 导致死锁：忘记关闭 jobs channel。如果所有任务都已分发但 channel 未关闭，worker 会一直阻塞在 <-jobs，等待一个永远不会到来的新任务，导致 wg.Wait() 死锁。
向已关闭的 channel 发送数据：在 jobs channel 关闭后，如果还有 goroutine 试图向其发送任务，会引发 panic。确保分发逻辑在 channel 关闭前完成。
错误处理被忽略：在 worker 函数中，没有正确处理任务执行过程中返回的 error，导致错误被静默丢失。

6. 小结

Worker Pool 模式是控制并发、保护有限资源的核心模式，通过复用一组固定的 goroutine 来处理大量任务。
核心组件包括：任务队列（channel）、一组 Worker goroutine、以及分发和协调逻辑。
带缓冲的 channel 是实现任务队列的天然选择，同时也可作为信号量来精确控制并发度。
优雅关闭依赖于 sync.WaitGroup 和 关闭 channel 的组合。WaitGroup 确保等待所有 worker 完成当前任务，而关闭 channel 则是通知它们“没有新任务了”的标准方式。
一个健壮的实现必须妥善处理错误和超时，并确保所有资源被正确清理，避免死锁和 panic。

并发模式：Worker Pool