Go并发模型中的Worker Pool模式
Go语言的goroutine轻量级特性使并发编程门槛大幅降低，但无限制创建goroutine会导致内存暴涨和调度器压力激增。Worker Pool通过固定数量的worker goroutine消费任务队列，将并发度控制在可预测范围内，避免资源耗尽。在高并发后端服务中，Worker Pool是处理批量IO操作（HTTP请求、数据库查询、文件读写）的标准范式。
Worker Pool的核心结构：一个带缓冲的task channel、一组worker goroutine、一个result channel。任务发送方将任务写入task channel，worker从channel取任务执行，结果写入result channel。
package workerpool
import "sync"
type Task struct {
ID int
Data interface{}
}
type Result struct {
TaskID int
Value interface{}
Err error
}
type Pool struct {
tasks chan Task
results chan Result
workers int
handler func(Task) (interface{}, error)
wg sync.WaitGroup
}
func NewPool(workers, bufferSize int, handler func(Task) (interface{}, error)) *Pool {
return &Pool{
tasks: make(chan Task, bufferSize),
results: make(chan Result, bufferSize),
workers: workers,
handler: handler,
}
}
func (p *Pool) Start() {
for i := 0; i < p.workers; i++ {
p.wg.Add(1)
go func(workerID int) {
defer p.wg.Done()
for task := range p.tasks {
value, err := p.handler(task)
p.results <- Result{TaskID: task.ID, Value: value, Err: err}
}
}(i)
}
}
func (p *Pool) Submit(task Task) {
p.tasks <- task
}
func (p *Pool) Close() {
close(p.tasks)
p.wg.Wait()
close(p.results)
}
func (p *Pool) Results() <-chan Result {
return p.results
}Fan-Out/Fan-In模式原理与实现
Fan-Out指将一个任务分发给多个goroutine并行处理，Fan-In指将多个goroutine的结果汇聚到一个channel。Fan-Out/Fan-In适合可并行拆分的计算密集型任务，如图片处理、数据聚合、批量API调用。
package fan
import "sync"
// Fan-Out: 将输入channel复制到n个输出channel
func FanOut[T any](in <-chan T, n int) []<-chan T {
channels := make([]<-chan T, n)
for i := 0; i < n; i++ {
ch := make(chan T)
channels[i] = ch
go func(c chan<- T) {
defer close(c)
for item := range in {
c <- item
}
}(ch)
}
return channels
}
// Fan-In: 将多个channel合并到一个输出channel
func FanIn[T any](channels ...<-chan T) <-chan T {
out := make(chan T)
var wg sync.WaitGroup
wg.Add(len(channels))
for _, ch := range channels {
go func(c <-chan T) {
defer wg.Done()
for item := range c {
out <- item
}
}(ch)
}
go func() {
wg.Wait()
close(out)
}()
return out
}Go 1.18+的泛型使Fan-Out/Fan-In可用于任意类型，不再需要interface{}和类型断言。
组合模式：Fan-Out + Worker Pool实战
实际场景中Fan-Out和Worker Pool常组合使用。以下是一个批量URL抓取服务——Fan-Out分发URL到多个worker并行请求，Fan-In汇聚所有结果：
package main
import (
"net/http"
"sync"
"time"
)
type FetchResult struct {
URL string
StatusCode int
Duration time.Duration
Err error
}
func fetchWorker(url string) FetchResult {
start := time.Now()
resp, err := http.Get(url)
duration := time.Since(start)
result := FetchResult{URL: url, Duration: duration}
if err != nil {
result.Err = err
return result
}
defer resp.Body.Close()
result.StatusCode = resp.StatusCode
return result
}
func batchFetch(urls []string, concurrency int) []FetchResult {
tasks := make(chan string, len(urls))
results := make(chan FetchResult, len(urls))
var wg sync.WaitGroup
for i := 0; i < concurrency; i++ {
wg.Add(1)
go func() {
defer wg.Done()
for url := range tasks {
results <- fetchWorker(url)
}
}()
}
go func() {
for _, url := range urls {
tasks <- url
}
close(tasks)
}()
go func() {
wg.Wait()
close(results)
}()
var allResults []FetchResult
for r := range results {
allResults = append(allResults, r)
}
return allResults
}concurrency参数控制最大并发数，避免同时打开过多TCP连接导致端口耗尽或被目标服务器限流。生产环境中建议将concurrency设为50-100，配合http.Transport的MaxIdleConnsPerHost参数复用连接。
优雅关闭与context超时控制
生产级Worker Pool必须支持优雅关闭：收到终止信号后停止接收新任务、等待已提交任务完成、超时后强制退出。context.Context是Go中传播取消信号的标准机制：
func (p *Pool) StartWithContext(ctx context.Context) {
for i := 0; i < p.workers; i++ {
p.wg.Add(1)
go func() {
defer p.wg.Done()
for {
select {
case task, ok := <-p.tasks:
if !ok {
return
}
value, err := p.handler(task)
select {
case p.results <- Result{TaskID: task.ID, Value: value, Err: err}:
case <-ctx.Done():
return
}
case <-ctx.Done():
return
}
}
}()
}
}
// 使用示例：30秒超时
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
pool := NewPool(10, 100, handler)
pool.StartWithContext(ctx)select同时监听task channel和ctx.Done()，确保worker能及时响应取消信号。Fan-In侧同样需要在ctx.Done()时退出收集循环，避免goroutine泄漏。通过runtime.NumGoroutine()可在运行时监控goroutine数量，Worker Pool启动前后差值应等于workers数，若有泄漏说明某个goroutine未正确退出。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/go-yu-yan-bing-fa-mo-shi-workerpool-yu-fanoutfanin-shi-zhan/
