Go 有界工作池

示例说明

为每个任务开一个 goroutine 在十万级任务前会失控——调度与内存双双爆炸。有界池的模式是固定工人消费共享通道：生产端往通道投递，工人端循环接收，天然背压——通道满时生产端阻塞。退出协议用通道关闭：close 后 range 循环自然结束，工人全部退出后 WaitGroup 放行。

技术要点

• 固定工人数约束并发资源
• close 通道是标准的退出广播
• 每工人独立统计槽免锁聚合

代码解读

runPool 起四个工人，各自在 range 循环里处理任务并给自己的统计槽计数；分发协程把任务切片投入通道并监听取消。dispatchDone 等到分发完成再 close 通道，工人消化完存量后退出；stats 切片每个工人独占一个下标，无数据竞争。

使用建议

任务处理时长不均时可引入结果通道回收。关闭通道只能由唯一的发送方执行。worker 数按下游配额设定而非 CPU 核数。

代码示例

package main

import (
	"context"
	"fmt"
	"sync"
)

type Job struct {
	ID  int
	URL string
}

func runPool(ctx context.Context, jobs []Job, workers int) (map[int]int, error) {
	jobCh := make(chan Job)
	stats := make([]int, workers)
	var wg sync.WaitGroup

	for w := 0; w < workers; w++ {
		w := w
		wg.Add(1)
		go func() {
			defer wg.Done()
			for job := range jobCh {
				select {
				case <-ctx.Done():
					return // 优雅退出：剩余任务由调用方决定去留
				default:
					fmt.Println("worker", w, "handles", job.ID)
					stats[w]++
				}
			}
		}()
	}

	dispatchDone := make(chan struct{})
	go func() {
		defer close(dispatchDone)
		for _, job := range jobs {
			select {
			case jobCh <- job:
			case <-ctx.Done():
				return
			}
		}
	}()

	<-dispatchDone
	close(jobCh) // 关闭通道即全体工人的退出信号
	wg.Wait()

	perWorker := make(map[int]int, workers)
	for w, count := range stats {
		perWorker[w] = count
	}
	return perWorker, ctx.Err()
}

func main() {
	jobs := make([]Job, 12)
	for i := range jobs {
		jobs[i] = Job{ID: i, URL: fmt.Sprintf("https://example.com/%d", i)}
	}
	stats, err := runPool(context.Background(), jobs, 4)
	fmt.Println(stats, err)
}