Go语言并发模型构建在goroutine和channel之上，Worker Pool和Fan-Out/Fan-In是两种最常用的高阶并发模式。Worker Pool通过固定数量的goroutine处理任务队列，控制并发度防止资源耗尽；Fan-Out/Fan-In通过扇出分发任务、扇出汇聚结果，实现并行计算与结果收集的解耦。结合context取消、速率限制和错误传播，可以构建生产级高并发服务。
Go Worker Pool模式与context取消机制
Worker Pool的核心思想是预创建固定数量的worker goroutine，通过共享channel接收任务。任务生产者将任务写入任务channel，worker从channel读取并执行。这种模式有效控制并发goroutine数量，避免无限制创建goroutine导致的内存暴涨和调度开销。
context.Context贯穿整个生命周期管理，当外部信号触发取消时，所有worker应尽快退出。以下是支持优雅关闭的Worker Pool实现：
package main
import (
"context"
"fmt"
"sync"
"time"
)
type Task struct {
ID int
Input string
}
type Result struct {
TaskID int
Value string
Err error
}
func worker(ctx context.Context, id int, tasks <-chan Task, results chan<- Result, wg *sync.WaitGroup) {
defer wg.Done()
for {
select {
case <-ctx.Done():
fmt.Printf("worker %d shutting down\n", id)
return
case task, ok := <-tasks:
if !ok {
return
}
// 模拟任务处理
time.Sleep(100 * time.Millisecond)
select {
case results <- Result{TaskID: task.ID, Value: "processed:" + task.Input}:
case <-ctx.Done():
return
}
}
}
}
func NewWorkerPool(ctx context.Context, workerCount, queueSize int) (<-chan Result, chan<- Task) {
tasks := make(chan Task, queueSize)
results := make(chan Result, queueSize)
var wg sync.WaitGroup
for i := 0; i < workerCount; i++ {
wg.Add(1)
go worker(ctx, i+1, tasks, results, &wg)
}
// 监控goroutine：所有worker退出后关闭results channel
go func() {
wg.Wait()
close(results)
}()
return results, tasks
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
results, tasks := NewWorkerPool(ctx, 4, 100)
// 提交任务
go func() {
for i := 0; i < 20; i++ {
tasks <- Task{ID: i, Input: fmt.Sprintf("job-%d", i)}
}
close(tasks)
}()
// 收集结果
for r := range results {
fmt.Printf("result: task=%d value=%s err=%v\n", r.TaskID, r.Value, r.Err)
}
}这段实现的关键设计在于：任务channel和结果channel都设置了缓冲区，避免生产者和消费者之间的阻塞；select同时监听ctx.Done()和任务channel，确保取消信号能及时响应；监控goroutine通过WaitGroup等待所有worker退出后再关闭results channel，消费者通过range检测channel关闭自然退出。
Fan-Out/Fan-In并行计算模式实现
Fan-Out/Fan-In模式将一个任务拆分为多个子任务并行执行（Fan-Out），然后将各子任务的结果汇聚为最终结果（Fan-In）。与Worker Pool的区别在于，Fan-Out通常是一次性分发固定数量的子任务，而非持续接收任务队列。
package main
import (
"context"
"fmt"
"math/rand"
"sync"
"time"
)
// Fan-Out：将输入分散到多个goroutine并行处理
func fanOut[T any, R any](ctx context.Context, input []T, workerCount int, fn func(T) R) <-chan R {
out := make(chan R)
var wg sync.WaitGroup
wg.Add(workerCount)
// 任务分发channel
taskCh := make(chan T)
go func() {
defer close(taskCh)
for _, item := range input {
select {
case taskCh <- item:
case <-ctx.Done():
return
}
}
}()
// 启动多个worker并行消费任务
for i := 0; i < workerCount; i++ {
go func() {
defer wg.Done()
for item := range taskCh {
select {
case out <- fn(item):
case <-ctx.Done():
return
}
}
}()
}
// 所有worker完成后关闭输出channel
go func() {
wg.Wait()
close(out)
}()
return out
}
// Fan-In：汇聚多个channel的数据到一个channel
func fanIn[T any](ctx context.Context, 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 val := range c {
select {
case out <- val:
case <-ctx.Done():
return
}
}
}(ch)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
func main() {
ctx := context.Background()
input := make([]int, 100)
for i := range input {
input[i] = rand.Intn(100)
}
// Fan-Out：8个worker并行处理
results := fanOut(ctx, input, 8, func(n int) int {
time.Sleep(time.Duration(rand.Intn(50)) * time.Millisecond)
return n * n // 计算平方
})
// Fan-In：汇聚结果
total := 0
for r := range results {
total += r
}
fmt.Printf("sum of squares: %d\n", total)
}这段代码使用Go 1.18+的泛型参数，fanOut和fanIn可以处理任意类型的输入和输出。fanIn函数接收多个channel，为每个channel启动一个转发goroutine，将数据汇聚到统一的输出channel。多个fanOut的输出可以通过fanIn合并，实现多级并行流水线。
速率限制并发请求与rate.Limiter令牌桶
在调用外部API时，无限制的并发请求会触发限流或导致服务过载。golang.org/x/time/rate包提供基于令牌桶算法的速率限制器，可以精确控制请求发送速率。令牌桶以固定速率生成令牌，每次请求消耗一个令牌，桶满时丢弃多余令牌。
package main
import (
"context"
"fmt"
"net/http"
"sync"
"time"
"golang.org/x/time/rate"
)
type APIClient struct {
client *http.Client
limiter *rate.Limiter
}
func NewAPIClient(rps float64, burst int) *APIClient {
return &APIClient{
client: &http.Client{Timeout: 10 * time.Second},
// rate: 每秒令牌数，burst: 桶容量（允许瞬时突发）
limiter: rate.NewLimiter(rate.Limit(rps), burst),
}
}
func (c *APIClient) Do(ctx context.Context, req *http.Request) (*http.Response, error) {
// Wait阻塞直到获取令牌，或ctx取消
if err := c.limiter.Wait(ctx); err != nil {
return nil, fmt.Errorf("rate limiter: %w", err)
}
return c.client.Do(req.WithContext(ctx))
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
api := NewAPIClient(10, 5) // 每秒10个请求，突发5个
urls := []string{
"https://api.example.com/users",
"https://api.example.com/posts",
"https://api.example.com/comments",
"https://api.example.com/albums",
"https://api.example.com/photos",
}
var wg sync.WaitGroup
results := make(chan string, len(urls))
for _, url := range urls {
wg.Add(1)
go func(u string) {
defer wg.Done()
req, _ := http.NewRequest("GET", u, nil)
resp, err := api.Do(ctx, req)
if err != nil {
results <- fmt.Sprintf("error %s: %v", u, err)
return
}
defer resp.Body.Close()
results <- fmt.Sprintf("ok %s: %d", u, resp.StatusCode)
}(url)
}
go func() {
wg.Wait()
close(results)
}()
for r := range results {
fmt.Println(r)
}
}rate.Limiter的Wait方法在令牌不足时阻塞等待，而非直接返回错误，适合需要保证请求最终发送的场景。如果需要非阻塞语义，使用Allow()方法检查令牌可用性后立即返回。burst参数控制瞬时突发容量，设为1时退化为严格匀速模式，适合对速率极其敏感的服务。
errgroup错误传播与goroutine生命周期管理
原生sync.WaitGroup不支持错误传播，任一goroutine出错时无法通知其他goroutine退出。golang.org/x/sync/errgroup包扩展了WaitGroup，在任一goroutine返回错误时自动取消context，触发其他goroutine退出。errgroup还支持通过SetLimit设置最大并发数，内置信号量控制。
package main
import (
"context"
"fmt"
"net/http"
"time"
"golang.org/x/sync/errgroup"
)
func fetchURL(ctx context.Context, url string) error {
req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
if err != nil {
return fmt.Errorf("create request %s: %w", url, err)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
return fmt.Errorf("fetch %s: %w", url, err)
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
return fmt.Errorf("unexpected status %s: %d", url, resp.StatusCode)
}
fmt.Printf("fetched %s OK\n", url)
return nil
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
defer cancel()
// errgroup with context: 第一个错误会取消ctx
g, ctx := errgroup.WithContext(ctx)
// 限制最大并发数为3
g.SetLimit(3)
urls := []string{
"https://httpbin.org/get",
"https://httpbin.org/delay/2",
"https://httpbin.org/status/404", // 这个会返回错误
"https://httpbin.org/delay/3",
"https://httpbin.org/get",
}
for _, url := range urls {
// 注意：循环变量捕获问题，Go 1.22+已修复
url := url
g.Go(func() error {
return fetchURL(ctx, url)
})
}
if err := g.Wait(); err != nil {
fmt.Printf("group failed: %v\n", err)
} else {
fmt.Println("all succeeded")
}
}errgroup.WithContext创建一个可取消的context，当任一g.Go注册的函数返回非nil错误时，该context被取消。其他正在执行的goroutine通过检查ctx.Done()感知取消信号并提前退出，避免无意义的计算开销。g.SetLimit(3)将并发数限制为3，超出限制的g.Go调用会阻塞等待，效果等同于信号量但代码更简洁。
循环变量捕获是Go并发编程的经典陷阱。在Go 1.22之前，for循环中的变量在每次迭代间共享同一地址，goroutine闭包捕获的是最终值。解决方案是在循环体内创建局部副本（url := url）。Go 1.22修复了这个问题，循环变量每次迭代都是新变量，但理解这个机制对维护旧代码仍有必要。
综合模式：限流Worker Pool完整实现
将Worker Pool、速率限制和errgroup三者结合，可以构建生产级并发处理管道。以下实现展示了一个完整的限流Worker Pool，支持优雅关闭、错误传播和速率控制：
package main
import (
"context"
"fmt"
"sync"
"time"
"golang.org/x/sync/errgroup"
"golang.org/x/time/rate"
)
type Job struct {
ID int
Data string
}
type Processor struct {
workers int
queue int
limiter *rate.Limiter
}
func (p *Processor) Run(ctx context.Context, jobs []Job, process func(context.Context, Job) error) error {
g, ctx := errgroup.WithContext(ctx)
g.SetLimit(p.workers)
jobCh := make(chan Job, p.queue)
var once sync.Once
// 生产者
g.Go(func() error {
defer once.Do(func() { close(jobCh) })
for _, job := range jobs {
if err := p.limiter.Wait(ctx); err != nil {
return err
}
select {
case jobCh <- job:
case <-ctx.Done():
return ctx.Err()
}
}
return nil
})
// 消费者（多个worker通过g.SetLimit控制并发）
for i := 0; i < p.workers; i++ {
g.Go(func() error {
for job := range jobCh {
if err := process(ctx, job); err != nil {
return fmt.Errorf("job %d: %w", job.ID, err)
}
}
return nil
})
}
return g.Wait()
}
func main() {
p := &Processor{
workers: 5,
queue: 20,
limiter: rate.NewLimiter(rate.Limit(20), 5), // 20 QPS, burst 5
}
jobs := make([]Job, 50)
for i := range jobs {
jobs[i] = Job{ID: i, Data: fmt.Sprintf("data-%d", i)}
}
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
err := p.Run(ctx, jobs, func(ctx context.Context, job Job) error {
time.Sleep(50 * time.Millisecond)
fmt.Printf("processed job %d\n", job.ID)
return nil
})
if err != nil {
fmt.Printf("error: %v\n", err)
}
}这个综合实现中，errgroup同时管理生产者和消费者goroutine，任何一方的错误都会通过context取消传播到所有goroutine。生产者通过rate.Limiter控制任务投递速率，消费者通过g.SetLimit控制处理并发度。once.Do确保jobCh只关闭一次，避免重复关闭导致的panic。这种模式适用于需要限流的外部API批量调用、数据库批量写入等实际生产场景。
原创文章，作者：小编，如若转载，请注明出处：https://www.yunthe.com/go-yu-yan-bing-fa-mo-shi-workerpool-yu-fanoutfanin-shi-zhan/
