高级并发模式深入
Go 的并发哲学是：“不要通过共享内存来通信，而应该通过通信来共享内存。“本章深入讲解 Go 中几种核心并发模式的原理、实现和常见坑。
1. Fan-Out / Fan-In 模式
核心思想
- Fan-Out：一个输入 channel，分发到多个 worker 并行处理。
- Fan-In：多个输入 channel，合并到一个输出 channel。
🔬 实现
// Fan-Out
func fanOut(in <-chan int, workers int) []<-chan int {
chans := make([]<-chan int, workers)
for i := 0; i < workers; i++ {
ch := make(chan int)
chans[i] = ch
go func(out chan int) {
defer close(out)
for v := range in {
out <- v * 2 // 并行处理
}
}(ch)
}
return chans
}
// Fan-In
func fanIn(chans ...<-chan int) <-chan int {
out := make(chan int)
var wg sync.WaitGroup
wg.Add(len(chans))
for _, ch := range chans {
go func(c <-chan int) {
defer wg.Done()
for v := range c {
out <- v
}
}(ch)
}
go func() {
wg.Wait()
close(out)
}()
return out
}
// 使用
func main() {
input := make(chan int)
go func() {
for i := 1; i <= 100; i++ {
input <- i
}
close(input)
}()
workers := fanOut(input, 5)
results := fanIn(workers...)
for r := range results {
fmt.Println(r)
}
}🚨 陷阱：Fan-Out 模式下，如果某个 worker panic，只关闭它自己的输出 channel，不影响其他 worker。但 Fan-In 的
wg.Wait()+close(out)必须在单独的 goroutine 中执行，否则会死锁。
2. Pipeline 模式
核心思想
每个 stage 是一个函数：接收 channel，处理数据，返回 channel。多个 stage 串联形成流水线。
// Stage 1: 生成器
func generator(nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
out <- n
}
}()
return out
}
// Stage 2: 平方
func square(in <-chan int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
out <- n * n
}
}()
return out
}
// Stage 3: 过滤
func filter(in <-chan int, fn func(int) bool) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for n := range in {
if fn(n) {
out <- n
}
}
}()
return out
}
// 组装
func main() {
nums := generator(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
squared := square(nums)
result := filter(squared, func(n int) bool { return n > 50 })
for v := range result {
fmt.Println(v) // 64, 81, 100
}
}🔬 带错误处理的 Pipeline
type Result struct {
Value int
Err error
}
func safeSquare(in <-chan int) <-chan Result {
out := make(chan Result)
go func() {
defer close(out)
for n := range in {
out <- Result{Value: n * n}
}
}()
return out
}
// 消费端检查错误
for r := range resultChan {
if r.Err != nil {
log.Printf("error: %v", r.Err)
continue
}
process(r.Value)
}🚨 关键陷阱：Pipeline Stage 的退出
// 问题代码：如果消费者提前退出，上游 stage 永远阻塞
func consumer(in <-chan int) {
for i := 0; i < 3; i++ {
fmt.Println(<-in)
}
// 消费者退出，但上游 generator 还在往 channel 写
// → goroutine 泄漏
}
// 解决方案：传入 context 做取消
func generatorWithCtx(ctx context.Context, nums ...int) <-chan int {
out := make(chan int)
go func() {
defer close(out)
for _, n := range nums {
select {
case out <- n:
case <-ctx.Done():
return
}
}
}()
return out
}3. Worker Pool 模式
🔬 完整实现
type Job struct {
ID int
Payload interface{}
}
type Result struct {
JobID int
Data interface{}
Err error
}
type WorkerPool struct {
jobs chan Job
results chan Result
workers int
wg sync.WaitGroup
once sync.Once
}
func NewWorkerPool(workers, queueSize int) *WorkerPool {
return &WorkerPool{
jobs: make(chan Job, queueSize),
results: make(chan Result, queueSize),
workers: workers,
}
}
func (p *WorkerPool) Start(ctx context.Context) {
for i := 0; i < p.workers; i++ {
p.wg.Add(1)
go p.worker(ctx, i)
}
// 等待所有 worker 完成后关闭 results
go func() {
p.wg.Wait()
close(p.results)
}()
}
func (p *WorkerPool) worker(ctx context.Context, id int) {
defer p.wg.Done()
for {
select {
case <-ctx.Done():
return
case job, ok := <-p.jobs:
if !ok {
return
}
// 处理 job
data, err := process(job.Payload)
p.results <- Result{JobID: job.ID, Data: data, Err: err}
}
}
}
func (p *WorkerPool) Submit(job Job) {
p.jobs <- job
}
func (p *WorkerPool) Results() <-chan Result {
return p.results
}
func (p *WorkerPool) Close() {
p.once.Do(func() {
close(p.jobs)
})
}⚡ 性能提示：
- Worker 数量：CPU 密集型 =
runtime.GOMAXPROCS(0)；IO 密集型 = 可超过 CPU 核数。- Queue Size：太小会让 submitter 阻塞，太大会消耗过多内存。根据背压策略选择。
4. Future/Promise 模式
核心思想
Future 模式让你可以提前启动一个异步任务，稍后再获取其结果。在 Go 中，通常用 channel 或 sync.Once 来实现。
基础实现
// Future 封装一个异步计算，调用者可以在未来获取结果
type Future[T any] struct {
result T
err error
done chan struct{} // 关闭时表示计算完成
}
func NewFuture[T any](fn func() (T, error)) *Future[T] {
f := &Future[T]{
done: make(chan struct{}),
}
go func() {
f.result, f.err = fn()
close(f.done) // 通知等待者：计算完成
}()
return f
}
// Get 阻塞直到计算完成，返回结果
func (f *Future[T]) Get() (T, error) {
<-f.done
return f.result, f.err
}
// GetWithTimeout 带超时的获取
func (f *Future[T]) GetWithTimeout(timeout time.Duration) (T, error) {
select {
case <-f.done:
return f.result, f.err
case <-time.After(timeout):
var zero T
return zero, fmt.Errorf("future timeout after %v", timeout)
}
}
// 使用示例
func main() {
// 启动三个并行计算
userFuture := NewFuture(func() (string, error) {
return fetchUser(123) // 耗时操作
})
orderFuture := NewFuture(func() ([]Order, error) {
return fetchOrders(123) // 耗时操作
})
statsFuture := NewFuture(func() (Stats, error) {
return computeStats() // 耗时操作
})
// 等待所有结果
user, err := userFuture.Get()
orders, err := orderFuture.Get()
stats, err := statsFuture.Get()
}使用 channel 的简化版
// 更简单的实现：直接返回一个 channel（Go 惯用的 "channel as future"）
func asyncDoWork(input int) <-chan Result {
ch := make(chan Result, 1) // 带缓冲，避免 goroutine 泄漏
go func() {
defer close(ch)
result, err := expensiveComputation(input)
ch <- Result{Value: result, Err: err}
}()
return ch
}
// 使用
future := asyncDoWork(42) // 立即返回，计算在后台进行
// 做其他事情...
result := <-future // 需要结果时再等待💡 Future vs Pipeline vs Worker Pool：
- Future：启动少量异步任务，各自独立，后面需要各自的结果。适用场景：并行调用多个独立 API。
- Pipeline：数据流经多个 stage 处理。适用场景：ETL、数据流处理。
- Worker Pool：大量任务，固定数量的 worker 依次执行。适用场景：任务队列、批量处理。
并发 Future 的错误处理
// fan-in with error：启动多个 future，收集所有结果
func fanInFutures[T any](futures ...*Future[T]) ([]T, error) {
results := make([]T, 0, len(futures))
for _, f := range futures {
result, err := f.Get()
if err != nil {
return nil, err // 快速失败
}
results = append(results, result)
}
return results, nil
}4. 限流器 (Rate Limiter)
令牌桶实现
type RateLimiter struct {
tokens chan struct{}
ticker *time.Ticker
done chan struct{}
once sync.Once
}
func NewRateLimiter(ratePerSecond int, burst int) *RateLimiter {
rl := &RateLimiter{
tokens: make(chan struct{}, burst),
ticker: time.NewTicker(time.Second / time.Duration(ratePerSecond)),
done: make(chan struct{}),
}
// 预填充 burst 个令牌
for i := 0; i < burst; i++ {
rl.tokens <- struct{}{}
}
// 定期补充令牌
go func() {
for {
select {
case <-rl.ticker.C:
select {
case rl.tokens <- struct{}{}:
default:
// 桶满了，丢弃令牌
}
case <-rl.done:
return
}
}
}()
return rl
}
func (rl *RateLimiter) Allow() bool {
select {
case <-rl.tokens:
return true
default:
return false
}
}
func (rl *RateLimiter) Wait(ctx context.Context) error {
select {
case <-rl.tokens:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (rl *RateLimiter) Close() {
rl.once.Do(func() {
close(rl.done)
rl.ticker.Stop()
})
}5. 🔬 并发陷阱深度解析
5.1 死锁的四要素
- 互斥：资源一次只能被一个 goroutine 使用。
- 持有并等待：goroutine 持有一个资源，等待另一个。
- 不可剥夺：资源只能由持有者释放。
- 循环等待：A 等 B，B 等 A。
// 经典死锁：两个 goroutine 以相反顺序获取两个锁
func deadlock() {
var mu1, mu2 sync.Mutex
go func() {
mu1.Lock()
time.Sleep(time.Millisecond)
mu2.Lock() // 等待 mu2
defer mu2.Unlock()
defer mu1.Unlock()
}()
go func() {
mu2.Lock()
time.Sleep(time.Millisecond)
mu1.Lock() // 等待 mu1 → 死锁
defer mu1.Unlock()
defer mu2.Unlock()
}()
}
// 解决：固定锁顺序
// 始终先锁 mu1 再锁 mu2（或使用 TryLock 做回退）✅ 正确做法 1：固定加锁顺序
// 所有的 goroutine 都先锁 mu1、再锁 mu2
func noDeadlock() {
var mu1, mu2 sync.Mutex
go func() {
mu1.Lock()
defer mu1.Unlock()
time.Sleep(time.Millisecond)
mu2.Lock()
defer mu2.Unlock()
}()
go func() {
mu1.Lock() // 也是先锁 mu1！
defer mu1.Unlock()
time.Sleep(time.Millisecond)
mu2.Lock() // 再锁 mu2
defer mu2.Unlock()
}()
}✅ 正确做法 2：一次性获取所有锁
// 将多个锁合并为一个锁
type SafeData struct {
mu sync.Mutex
a, b int
}✅ 正确做法 3：使用 sync.TryLock（Go 1.18+）做回退
go func() {
mu1.Lock()
defer mu1.Unlock()
if !mu2.TryLock() {
// 获取不到 mu2，先释放 mu1 再重试
mu1.Unlock()
time.Sleep(time.Millisecond)
// 或重新尝试
return
}
defer mu2.Unlock()
}()5.2 Goroutine 泄漏
// 泄漏场景 1：channel 永远没有接收者
func leak1() {
ch := make(chan int)
go func() {
ch <- 42 // 阻塞，goroutine 永远挂在这里
}()
// ch 被丢弃了…
}
// 泄漏场景 2：context 被忽略
func leak2(ctx context.Context) {
for {
select {
case <-time.After(time.Second):
// 没有检查 ctx.Done()，无法退出
doWork()
}
}
}
// 检测：runtime.NumGoroutine()，pprof 的 goroutine profile
// ✅ 泄漏场景 1 的正确做法：给 channel 提供接收者或使用带缓冲的 channel
func fixLeak1() {
ch := make(chan int, 1) // 带缓冲，发送不阻塞
go func() {
ch <- 42 // 不阻塞（缓冲区有空位）
}()
// 即使 ch 被丢弃，goroutine 最终也会结束
}
// ✅ 泄漏场景 2 的正确做法：添加 ctx.Done() 分支
func fixLeak2(ctx context.Context) {
timer := time.NewTimer(time.Second)
defer timer.Stop()
for {
select {
case <-timer.C:
doWork()
timer.Reset(time.Second)
case <-ctx.Done():
return // 可以正常退出
}
}
}5.3 Channel 关闭的规则
| 操作 | nil chan | 已关闭 chan | 正常 chan |
|---|---|---|---|
ch <- v |
永久阻塞 | panic | 阻塞或成功 |
<-ch |
永久阻塞 | 立即返回零值 | 阻塞或成功 |
close(ch) |
panic | panic | 成功 |
💡 黄金法则：发送方负责关闭 channel；接收方不应关闭。
5.4 使用 -race 检测竞态条件
go test -race ./...
go run -race main.go
go build -race -o myapp-race 会检测：
- 无同步的并发读写同一内存
WaitGroup.Done计数错误- 已关闭 channel 的 send
⚡
-race会使程序慢 ~10 倍、内存增加 ~5-10 倍，仅用于开发和测试。
6. 并发模式选择指南
| 场景 | 推荐方案 |
|---|---|
| 简单共享数据保护 | sync.Mutex |
| 读多写少 | sync.RWMutex |
| goroutine 间的数据流 | channel |
| 等待多个 goroutine 完成 | sync.WaitGroup |
| 确保只执行一次 | sync.Once |
| 等待条件成立 | sync.Cond 或 channel + close |
| 计数/状态标志 | sync/atomic |
| 任务分发+结果收集 | Worker Pool |
| 多阶段数据处理 | Pipeline |