sync.WaitGroup、Mutex 和 errgroup：Go 中的 goroutine 协调

1. 概述

在 Go 中启动一个 goroutine 很容易，只需添加 go 关键字。困难的部分是知道何时 goroutine 完成、保护共享数据以及处理 goroutine 失败时的错误。

本文介绍这三个工具：sync.WaitGroup、sync.Mutex 和 errgroup.Group。读完后，你将知道在不同情况下选择哪个工具、如何编写使用 -race 运行的测试以及要避免的常见错误。

示例使用 Go 1.22 及更高版本。从这个版本开始，每个 for 循环为每次迭代创建新变量，所以你不再需要写 id := id。

2. 示例问题

假设你有一个订单编号列表。每个订单需要调用一个缓慢的服务，耗时约 100ms。如果按顺序调用 10 个订单，你需要花大约 1 秒。如果并行调用，你只需花约 100ms。

这是模拟服务调用的函数：

package main

import (
	"fmt"
	"time"
)

func fetchOrder(id int) string {
	time.Sleep(100 * time.Millisecond)
	return fmt.Sprintf("order-%d", id)
}

如果你只在循环中写 go fetchOrder(id)，main 函数会在 goroutine 完成之前结束。程序退出，结果丢失。下一节处理这个问题。

3. sync.WaitGroup：等待一组 goroutine

3.1. 基本用法

WaitGroup 是一个计数器。你用 Add 增加计数，用 Done 减少计数，Wait 阻塞直到计数变为 0。

func main() {
	var wg sync.WaitGroup
	ids := []int{1, 2, 3}

	for _, id := range ids {
		wg.Add(1)
		go func() {
			defer wg.Done()
			fmt.Println("fetched", fetchOrder(id))
		}()
	}

	wg.Wait()
	fmt.Println("all done")
}

上面的代码工作如下：

• wg.Add(1) 在启动 goroutine 之前 被调用，在父 goroutine 中。
• defer wg.Done() 确保计数总是被减少，即使 goroutine 提前结束或 panic。
• wg.Wait() 阻塞 main 直到三个 goroutine 都调用了 Done。

结果可能如下所示。fetched 行的顺序在不同运行之间会变化：

fetched order-2
fetched order-1
fetched order-3
all done

3.2. Go 1.25 中的 WaitGroup.Go

从 Go 1.25 开始，WaitGroup 增加了 Go 方法。这个方法将 Add、go 和 Done 合并为一个调用：

var wg sync.WaitGroup
for _, id := range ids {
	wg.Go(func() {
		fmt.Println("fetched", fetchOrder(id))
	})
}
wg.Wait()

这样写更简洁，避免了两个常见错误：忘记调用 Done，或在 goroutine 内调用 Add。

3.3. 无需 lock 地收集结果

你通常需要保留结果，而不仅仅是打印。一个安全的方法是创建预分配长度的切片，然后让每个 goroutine 写入自己的索引：

results := make([]string, len(ids))
var wg sync.WaitGroup
for i, id := range ids {
	wg.Go(func() {
		results[i] = fetchOrder(id)
	})
}
wg.Wait()
fmt.Println(results)

[order-1 order-2 order-3]

没有 goroutine 写入同一个元素，所以这里没有数据竞争。wg.Wait() 还建立了 happens-before 关系。因此，在 Wait 之后，父 goroutine 可以读取子 goroutine 写入的完整数据。

注意： 不要在 goroutine 内部调用 wg.Add(1)。如果这样做，Wait() 可能在 Add 之前运行并立即返回，而工作尚未完成。

4. sync.Mutex：保护共享数据

4.1. 数据竞争是什么样的

现在假设你需要按客户累计收入，多个 goroutine 同时更新一个 map：

revenue := map[string]float64{}
var wg sync.WaitGroup
for i := 0; i < 1000; i++ {
	wg.Go(func() {
		revenue["alice"] += 10
	})
}
wg.Wait()

这个程序可能会崩溃并出现以下错误：

fatal error: concurrent map writes

只用一个 int 变量的情况更危险。程序不会崩溃，而是悄悄产生错误的结果，例如 9730 而不是 10000。要检测这种错误，请使用 race detector 运行：

go run -race main.go

WARNING: DATA RACE
Write at 0x00c000012345 by goroutine 8:
...

4.2. 用 Mutex 修复

sync.Mutex 确保在任何时刻只有一个 goroutine 执行 Lock 和 Unlock 之间的代码。通常的做法是将 mutex 与它保护的数据一起包装在同一个 struct 中：

package shop

import "sync"

type RevenueTracker struct {
	mu    sync.Mutex
	total map[string]float64
}

func NewRevenueTracker() *RevenueTracker {
	return &RevenueTracker{total: make(map[string]float64)}
}

func (r *RevenueTracker) Add(customer string, amount float64) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.total[customer] += amount
}

func (r *RevenueTracker) Total(customer string) float64 {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.total[customer]
}

需要注意的几点：

• 对 total 的每次访问，包括读，都必须通过 lock。
• defer r.mu.Unlock() 确保 lock 总是被释放，即使函数有多个 return 分支。
• 方法使用指针接收器 (*RevenueTracker)。如果使用值接收器，每次调用都会复制 mutex，该 lock 就无法保护任何内容。

当读操作远多于写操作时，你可以使用 sync.RWMutex。读方法调用 RLock/RUnlock，写方法调用 Lock/Unlock。多个 goroutine 可以同时读，但写时只有一个 goroutine 能持有 lock。

注意： 不要复制一个已经使用过的包含 Mutex 的 struct。go vet 命令会用 copylocks 警告这种情况。另外，Go 没有可重入锁：如果一个 goroutine 在同一个 mutex 上调用 Lock() 两次，它会发生死锁。

4.3. 何时不需要 Mutex

对于简单的计数器，sync/atomic 更简洁高效：

var processed atomic.Int64
processed.Add(1)
fmt.Println(processed.Load())

当数据需要以流的形式从一个 goroutine 流向另一个时，channel 通常更合适。当你有共享状态（如 map 或 struct），多个 goroutine 需要读写它时，Mutex 最合适。

5. errgroup：WaitGroup 的错误处理版本

WaitGroup 不知道错误。如果 fetchOrder 可以返回 error，你必须自己用 mutex 或 channel 收集错误，并自己取消剩余的 goroutine。errgroup 包已经处理了这些事情。

5.1. 添加依赖

errgroup 位于 golang.org/x/sync 模块中，不属于标准库：

go get golang.org/x/sync

5.2. 汇总错误并通过 context 取消

第一步是重写 fetchOrder 使其接收 context.Context 并可以返回错误：

package shop

import (
	"context"
	"fmt"
	"time"

	"golang.org/x/sync/errgroup"
)

type Order struct {
	ID       int
	Customer string
}

func fetchOrder(ctx context.Context, id int) (Order, error) {
	if id < 0 {
		return Order{}, fmt.Errorf("invalid order id %d", id)
	}
	select {
	case <-time.After(100 * time.Millisecond):
		return Order{ID: id, Customer: "alice"}, nil
	case <-ctx.Done():
		return Order{}, ctx.Err()
	}
}

func LoadOrders(ctx context.Context, ids []int) ([]Order, error) {
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(3)

	orders := make([]Order, len(ids))
	for i, id := range ids {
		g.Go(func() error {
			o, err := fetchOrder(ctx, id)
			if err != nil {
				return err
			}
			orders[i] = o
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}
	return orders, nil
}

上面的代码工作如下：

• errgroup.WithContext(ctx) 创建一个 group 和一个子 context。当一个 goroutine 返回第一个错误时，子 context 立即被取消。
• g.SetLimit(3) 允许最多 3 个 goroutine 同时运行。下一次调用 g.Go 将阻塞直到有空位。这样当列表有数千个订单时，你就不会让后端服务过载。
• g.Go 接收一个类型为 func() error 的函数，你不需要调用 Add/Done。
• g.Wait() 等待所有 goroutine 完成，然后返回第一个错误。之后发生的错误会被忽略。
• fetchOrder 监听 ctx.Done()，所以剩余的 goroutine 会提前停止，而不是运行完整的 100ms。

用无效的订单编号调用函数：

_, err := shop.LoadOrders(context.Background(), []int{1, 2, -5, 4})
fmt.Println(err)

invalid order id -5

如果你需要尝试运行但达到限制时不阻塞，请使用 g.TryGo(f)。这个函数返回 false 而不是等待。

注意： 取消 context 只在 goroutine 内的代码主动检查 ctx 时有效。如果 fetchOrder 使用 time.Sleep 但不监听 context，即使 group 已失败，goroutine 仍会运行完整时间。

6. 使用 -race 编写测试

Go 没有 JUnit 或 AssertJ。你使用标准库的 testing 包。最重要的是始终使用 -race 标志运行测试，因为许多并发错误只有在有 race detector 时才会显露。

package shop

import (
	"context"
	"sync"
	"testing"
)

func TestRevenueTracker_ConcurrentAdd(t *testing.T) {
	tracker := NewRevenueTracker()
	var wg sync.WaitGroup

	for i := 0; i < 1000; i++ {
		wg.Go(func() {
			tracker.Add("alice", 10)
		})
	}
	wg.Wait()

	if got := tracker.Total("alice"); got != 10000 {
		t.Fatalf("Total(alice) = %v, want 10000", got)
	}
}

func TestLoadOrders_ReturnsAllOrders(t *testing.T) {
	orders, err := LoadOrders(context.Background(), []int{1, 2, 3})
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	for i, o := range orders {
		if o.ID != i+1 {
			t.Errorf("orders[%d].ID = %d, want %d", i, o.ID, i+1)
		}
	}
}

func TestLoadOrders_StopsOnFirstError(t *testing.T) {
	orders, err := LoadOrders(context.Background(), []int{1, -1, 3})
	if err == nil {
		t.Fatal("expected error, got nil")
	}
	if orders != nil {
		t.Errorf("expected nil orders, got %v", orders)
	}
}

运行测试：

go test -race ./...

ok  	example.com/shop	0.412s

尝试删除 Add 中的两行 Lock/Unlock 然后重新运行。测试会失败并显示 WARNING: DATA RACE 或因 concurrent map writes 而崩溃。这是一种快速检查 mutex 是否真正必要的方法。

第二个测试也确认结果的顺序被正确保留：orders[i] 与 ids[i] 匹配，尽管 goroutine 以任意顺序完成。

7. 常见错误和工具选择

最常见的错误：

• 在 goroutine 中调用 Add。 Wait 可能返回过早。在 go 命令之前调用 Add，或使用 wg.Go。
• 按值传递 WaitGroup 或 Mutex。 函数收到一份副本，原始计数器或 lock 不会改变。按指针 *sync.WaitGroup 传递，或使用闭包。
• 在执行 I/O 时持有 lock。 例如在 Lock 内调用 HTTP。所有其他 goroutine 都必须排队等待，程序失去并行运行的效果。只在最短的代码段内持有 lock。
• 使用空的 errgroup.Group（var g errgroup.Group）但期望 context 被取消。 只有 WithContext 才创建可取消的 context。
• 忽视 g.Wait() 的返回值。 这样错误会无声地消失。
• 为成千上万的请求无限启动 goroutine。 使用 SetLimit 或 worker pool。

工具选择：

情景	工具
等待多个不返回错误的任务	sync.WaitGroup
多个 goroutine 修改共享的 map 或 struct	sync.Mutex / sync.RWMutex
简单的计数器或标志	sync/atomic
任务可能出错、需要提前取消、限制 goroutine 数量	errgroup

注意： errgroup 只保留第一个错误。如果你需要知道所有错误，例如报告哪些订单出错，让每个 goroutine 将错误写入 errs[i]，然后在调用 Wait 后用 errors.Join(errs...) 合并。

8. 总结

sync.WaitGroup 回答问题“何时完成？”。sync.Mutex 回答问题“现在谁在修改数据？”。errgroup 结合等待、错误处理、context 取消和 goroutine 限制。

在实际代码中，当并行调用多个服务时，errgroup.WithContext 配合 SetLimit 通常是合理的默认选择。无论使用哪个工具，始终运行 go test -race。这是在数据竞争进入生产前捕获它们的最便宜的方法。