1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

Worker Pool 的 goroutine 与 channel 拓扑

  main goroutine                    workers（固定 N 个）          收集方
  ──────────────                    ────────────────────          ──────
                                   ┌──────────────┐
                    ┌──► close(in) │ worker 1     │──┐
  for _, j := range │              │ for j := range in │  │
      jobs          │              └──────────────┘  │
        │           │                                │
        └─► in ─────┤              ┌──────────────┐  ├──► out ──► for r := range out
            (无缓冲) │              │ worker 2     │  │   (缓冲 = len(jobs))
                    └─────────────►│ ...          │──┘
                                   └──────────────┘
                                   ┌──────────────┐
                                   │ worker N     │──┘
                                   └──────────────┘
                                        │
                                        ▼
                              wg.Wait() 之后 close(out) 🔥
                              └─► 收集方的 range 才能正常结束

  三个必须存在的收尾动作：
   ① 生产者 defer close(in)     → worker 的 range 能结束
   ② wg.Wait() 后 close(out)    → 收集方的 range 能结束
   ③ 每个 select 都有 ctx.Done() → 取消时所有 goroutine 能退出

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

func RunPool(ctx context.Context, jobs []Job, workers int) ([]Result, error) {
	in := make(chan Job)
	out := make(chan Result, len(jobs))

	var wg sync.WaitGroup
	for range workers {                       // 🆕 1.25：wg.Go 一步到位
		wg.Go(func() {
			for j := range in {
				select {
				case out <- process(j):
				case <-ctx.Done():
					return                 // 🔥 每个 worker 都要能退出
				}
			}
		})
	}

	go func() {
		defer close(in)                    // 🔥 生产者负责关闭
		for _, j := range jobs {
			select {
			case in <- j:
			case <-ctx.Done():
				return
			}
		}
	}()

	go func() { wg.Wait(); close(out) }()  // 🔥 全部 worker 结束后关闭 out

	var results []Result
	for r := range out {                   // range 会在 out 关闭后自动结束
		results = append(results, r)
	}
	return results, ctx.Err()
}

1

pool: [2 4 6]         # 输入 [1 2 3]，2 个 worker，各自乘 2

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

func gen(ctx context.Context, ns ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)                  // 🔥 每级都要 close 自己的输出
		for _, n := range ns {
			select {
			case out <- n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

func double(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for v := range in {               // 上游关闭 → 循环自动结束
			select {
			case out <- v * 2:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

for v := range double(ctx, gen(ctx, 1, 2, 3)) {
	fmt.Println(v)                        // 2 4 6
}

1

pipeline: [2 4 6]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

func merge(ctx context.Context, chans ...<-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup

	for _, c := range chans {
		wg.Go(func() {                    // ⚠️ wg.Go 已含 Done，别再手动调
			for v := range c {
				select {
				case out <- v:
				case <-ctx.Done():
					return
				}
			}
		})
	}

	go func() { wg.Wait(); close(out) }()  // 🔥 全部合并完再关闭
	return out
}

// 使用：把一个大任务切成 3 份并行处理
c1 := process(ctx, part1)
c2 := process(ctx, part2)
c3 := process(ctx, part3)
for v := range merge(ctx, c1, c2, c3) {
	// 输出顺序不确定 ⚠️ 需要有序就带上序号再排序
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

type indexed struct {
	idx int
	val Result
}

results := make([]Result, len(inputs))
var wg sync.WaitGroup
for i, in := range inputs {
	wg.Go(func() {
		results[i] = process(in)          // 🔥 各写各的下标，无需锁
	})
}
wg.Wait()
// results 天然有序 ✅

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

func fetchWithTimeout(ctx context.Context, url string) ([]byte, error) {
	ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()                        // 🔥 必须，否则泄漏 timer

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return nil, fmt.Errorf("new request: %w", err)
	}
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("get %s: %w", url, err)   // ctx 错误已在链里
	}
	defer resp.Body.Close()
	return io.ReadAll(resp.Body)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

func retry(ctx context.Context, attempts int, f func(context.Context) error) error {
	var err error
	backoff := 100 * time.Millisecond

	for i := range attempts {
		if err = f(ctx); err == nil {
			return nil
		}
		if ctx.Err() != nil {             // 上下文已取消，没必要再试
			return ctx.Err()
		}
		if !isRetryable(err) {            // 🔥 不可重试的错误立即返回
			return err
		}

		timer := time.NewTimer(backoff)
		select {
		case <-ctx.Done():
			timer.Stop()
			return ctx.Err()
		case <-timer.C:
		}
		backoff *= 2                      // 指数退避
		if backoff > 5*time.Second {
			backoff = 5 * time.Second     // 封顶
		}
		_ = i
	}
	return fmt.Errorf("after %d attempts: %w", attempts, err)
}

func isRetryable(err error) bool {
	// 🔥 只重试「暂时性」错误；参数错误、404 之类重试也没用
	return !errors.Is(err, ErrInvalidInput) && !errors.Is(err, ErrNotFound)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

func main() {
	// ① 信号 → ctx（收到 SIGINT/SIGTERM 时取消）🔥
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	srv := &http.Server{Addr: ":8080", Handler: mux}

	go func() {
		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			log.Fatalf("listen: %v", err)
		}
	}()

	// ② 后台任务用同一个 ctx
	var wg sync.WaitGroup
	wg.Go(func() { runWorker(ctx) })

	// ③ 等信号
	<-ctx.Done()
	log.Println("shutting down...")
	stop()                                 // 恢复默认信号行为，第二次 Ctrl-C 强制退出

	// ④ 给在途请求一个有限的完成窗口
	shutdownCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
	defer cancel()
	if err := srv.Shutdown(shutdownCtx); err != nil {
		log.Printf("forced shutdown: %v", err)
	}

	wg.Wait()                              // ⑤ 等后台任务退出
	log.Println("bye")
}

1
2
3
4

triggering shutdown...
  client got: 200              ← 在途请求被等完，正常返回 ✅
shutdown returned, waiting workers
clean exit

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

func fetchAll(ctx context.Context, urls []string, limit int) ([]string, error) {
	sem := make(chan struct{}, limit)      // 最多 limit 个并发
	results := make([]string, len(urls))

	g, ctx := errgroup.WithContext(ctx)
	for i, u := range urls {
		g.Go(func() error {
			select {
			case sem <- struct{}{}:        // 获取令牌（满了就等）
			case <-ctx.Done():
				return ctx.Err()
			}
			defer func() { <-sem }()       // 🔥 释放令牌

			body, err := fetch(ctx, u)
			if err != nil {
				return fmt.Errorf("fetch %s: %w", u, err)
			}
			results[i] = body
			return nil
		})
	}
	if err := g.Wait(); err != nil {
		return nil, err
	}
	return results, nil
}

1

peak concurrency: 3        # limit = 3，起 20 个任务

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// 队列满时阻塞生产者，而不是无限堆积内存
queue := make(chan Task, 100)

go func() {
	for t := range queue {
		handle(t)
	}
}()

for _, t := range tasks {
	queue <- t          // 队列满 → 生产者阻塞，形成背压 ✅
}
close(queue)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// 🛑 典型：提前 return，兄弟 goroutine 泄漏
func bad(ctx context.Context, items []Item) error {
	ch := make(chan result)               // 无缓冲
	for _, it := range items {
		go func() { ch <- process(it) }()  // 若下面提前 return，这些 goroutine 全部卡死
	}
	for range items {
		r := <-ch
		if r.err != nil {
			return r.err                  // 💥 剩余 goroutine 永远阻塞在 ch <- 上
		}
	}
	return nil
}

// ✅ 修复：加缓冲，或用 errgroup
func good(ctx context.Context, items []Item) error {
	ch := make(chan result, len(items))   // 🔥 缓冲兜底：即使无人接收也能写完
	...
}

1
2
3
4
5
6
7
8
9

// 测试里做泄漏断言
func TestNoLeak(t *testing.T) {
	before := runtime.NumGoroutine()
	doWork()
	time.Sleep(50 * time.Millisecond)     // 给 goroutine 时间退出
	if after := runtime.NumGoroutine(); after > before {
		t.Errorf("goroutine leak: %d -> %d", before, after)
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

func Process(ctx context.Context, inputs []Input, workers, limit int) ([]Result, error) {
	g, ctx := errgroup.WithContext(ctx)        // ① 一处失败，全局取消
	g.SetLimit(limit)                          // ② 并发上限

	results := make([]Result, len(inputs))

	for i, in := range inputs {
		g.Go(func() error {
			// ③ 每个任务都要尊重取消
			if err := ctx.Err(); err != nil {
				return err
			}

			r, err := processOne(ctx, in)
			if err != nil {
				return fmt.Errorf("process %d: %w", i, err)   // ④ 带上下文包装
			}

			results[i] = r                     // ⑤ 按下标写，无需锁，天然有序
			return nil
		})
	}

	if err := g.Wait(); err != nil {           // ⑥ 等全部完成或首个错误
		return nil, err
	}
	return results, nil
}