一、并发设计模式概览
1.1 为什么需要并发模式
并发编程的三大挑战 ┌─────────────────────────────────────────────────────────────┐ │ 1. 竞态条件 (Race Condition) │ │ 多个 goroutine 同时读写共享数据 │ │ 解决方案: mutex / channel / atomic │ │ │ │ 2. 死锁 (Deadlock) │ │ goroutine 互相等待对方释放资源 │ │ 解决方案: 按序加锁 / timeout / 锁粒度控制 │ │ │ │ 3. 资源泄漏 (Resource Leak) │ │ goroutine 永远无法退出 │ │ 解决方案: context 取消 / 超时控制 / 优雅关闭 │ └─────────────────────────────────────────────────────────────┘ Go 并发哲学: "Don't communicate by sharing memory; share memory by communicating." "不要通过共享内存来通信，而要通过通信来共享内存。"1.2 核心原语回顾
// 1. goroutine: 轻量级线程 go func() { // 并发执行 }() // 2. channel: 通信机制 ch := make(chan int) // 无缓冲 channel ch := make(chan int, 10) // 有缓冲 channel ch <- value // 发送 value := <-ch // 接收 // 3. select: 多路复用 select { case v := <-ch1: // 处理 ch1 case v := <-ch2: // 处理 ch2 case <-time.After(1 * time.Second): // 超时处理 default: // 非阻塞操作 } // 4. sync 包 var mu sync.Mutex // 互斥锁 var rw sync.RWMutex // 读写锁 var wg sync.WaitGroup // 等待组 var once sync.Once // 单次执行二、Fan-In 模式（扇入）
2.1 概念与实现
Fan-In 模式 ┌─────────────────────────────────────────────────────────────┐ │ 多个输入 channel → 合并到一个输出 channel │ │ │ │ ┌─────────┐ │ │ │ ch1 ────┤ │ │ ├─────────┤ │ │ │ ch2 ────┼──→ mergedCh │ │ ├─────────┤ │ │ │ ch3 ────┤ │ │ └─────────┘ │ │ │ │ 适用场景: │ │ - 聚合多个数据源的结果 │ │ - 合并多个 worker 的输出 │ │ - 日志收集系统 │ └─────────────────────────────────────────────────────────────┘// fan/fan_in.go package main import ( "fmt" "math/rand" "sync" "time" ) // 基础 Fan-In 实现 func fanIn(channels ...<-chan int) <-chan int { merged := make(chan int) var wg sync.WaitGroup // 为每个输入 channel 启动一个 goroutine for _, ch := range channels { wg.Add(1) go func(c <-chan int) { defer wg.Done() for v := range c { merged <- v } }(ch) } // 等待所有 goroutine 完成后关闭 merged go func() { wg.Wait() close(merged) }() return merged } // 带超时的 Fan-In func fanInWithTimeout(timeout time.Duration, channels ...<-chan int) (<-chan int, <-chan struct{}) { merged := make(chan int) done := make(chan struct{}) var wg sync.WaitGroup for _, ch := range channels { wg.Add(1) go func(c <-chan int) { defer wg.Done() for { select { case v, ok := <-c: if !ok { return } merged <- v case <-time.After(timeout): return // 超时退出 } } }(ch) } go func() { wg.Wait() close(merged) close(done) }() return merged, done } // 示例：从多个数据源聚合 func dataSource(name string, count int) <-chan int { ch := make(chan int) go func() { defer close(ch) for i := 0; i < count; i++ { ch <- rand.Intn(100) time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond) } }() return ch } func main() { // 创建多个数据源 source1 := dataSource("source1", 5) source2 := dataSource("source2", 3) source3 := dataSource("source3", 4) // 合并 merged := fanIn(source1, source2, source3) // 消费 for v := range merged { fmt.Printf("Received: %d\n", v) } }三、Fan-Out 模式（扇出）
3.1 概念与实现
Fan-Out 模式 ┌─────────────────────────────────────────────────────────────┐ │ 一个输入 channel → 分发到多个 worker │ │ │ │ ┌→ worker1 │ │ │ │ │ input ───┼→ worker2 │ │ │ │ │ └→ worker3 │ │ │ │ 适用场景: │ │ - 任务分发（负载均衡） │ │ - 并行处理大量请求 │ │ - 消息队列消费者 │ └─────────────────────────────────────────────────────────────┘// fan/fan_out.go package main import ( "fmt" "sync" "time" ) // Worker 函数类型 type Worker func(id int, jobs <-chan int, results chan<- int) // Fan-Out 实现 func fanOut(jobs <-chan int, numWorkers int, worker Worker) <-chan int { results := make(chan int) var wg sync.WaitGroup // 启动指定数量的 worker for i := 0; i < numWorkers; i++ { wg.Add(1) go func(id int) { defer wg.Done() worker(id, jobs, results) }(i) } // 等待所有 worker 完成 go func() { wg.Wait() close(results) }() return results } // 具体 Worker 实现 func processJob(id int, jobs <-chan int, results chan<- int) { for job := range jobs { // 模拟处理时间 time.Sleep(time.Duration(100+job*10) * time.Millisecond) result := job * 2 fmt.Printf("Worker %d processed job %d → %d\n", id, job, result) results <- result } } // 带错误处理的 Worker func processJobWithRetry(id int, jobs <-chan int, results chan<- int) { for job := range jobs { var result int var err error // 最多重试 3 次 for retry := 0; retry < 3; retry++ { result, err = riskyOperation(job) if err == nil { break } fmt.Printf("Worker %d retrying job %d (attempt %d)\n", id, job, retry+1) time.Sleep(time.Second) } if err != nil { fmt.Printf("Worker %d failed job %d: %v\n", id, job, err) continue } results <- result } } func riskyOperation(job int) (int, error) { // 模拟可能失败的操作 if job%3 == 0 { return 0, fmt.Errorf("random failure on job %d", job) } return job * 2, nil } func main() { // 创建任务 jobs := make(chan int, 20) go func() { for i := 1; i <= 10; i++ { jobs <- i } close(jobs) }() // 启动 3 个 worker results := fanOut(jobs, 3, processJob) // 收集结果 for result := range results { fmt.Printf("Result: %d\n", result) } }四、Pipeline 模式
4.1 概念与实现
Pipeline 模式 ┌─────────────────────────────────────────────────────────────┐ │ 一系列处理阶段，每个阶段通过 channel 连接 │ │ │ │ stage1 → stage2 → stage3 → stage4 │ │ ch1 ch2 ch3 │ │ │ │ 特点: │ │ - 每个阶段独立运行 │ │ - 通过 channel 解耦 │ │ - 天然支持并发 │ │ - 易于扩展和维护 │ │ │ │ 适用场景: │ │ - 数据处理流水线 │ │ - ETL 流程 │ │ - 图像/视频处理 │ │ - 日志处理系统 │ └─────────────────────────────────────────────────────────────┘// pipeline/pipeline.go package main import ( "fmt" "strings" "time" ) // 1. 生成阶段：产生原始数据 func generate(nums ...int) <-chan int { out := make(chan int) go func() { defer close(out) for _, n := range nums { out <- n time.Sleep(100 * time.Millisecond) } }() return out } // 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 } // 3. 加倍阶段：乘以 2 func double(in <-chan int) <-chan int { out := make(chan int) go func() { defer close(out) for n := range in { out <- n * 2 } }() return out } // 4. 过滤阶段：只保留偶数 func filterEven(in <-chan int) <-chan int { out := make(chan int) go func() { defer close(out) for n := range in { if n%2 == 0 { out <- n } } }() return out } // 5. 统计阶段：收集结果 func sink(in <-chan int) { var sum int count := 0 for n := range in { sum += n count++ fmt.Printf("Received: %d (running sum: %d)\n", n, sum) } fmt.Printf("Total: %d numbers, sum: %d\n", count, sum) } // 文本处理 Pipeline type TextPipeline struct { input chan string } func NewTextPipeline() *TextPipeline { return &TextPipeline{ input: make(chan string, 100), } } func (p *TextPipeline) Start() <-chan string { // 构建 pipeline // input → lowercase → splitWords → removeStopwords → output lowercased := p.lowercase() words := p.splitWords(lowercased) filtered := p.removeStopwords(words) return filtered } func (p *TextPipeline) Feed(text string) { p.input <- text } func (p *TextPipeline) Close() { close(p.input) } func (p *TextPipeline) lowercase() <-chan string { out := make(chan string) go func() { defer close(out) for text := range p.input { out <- strings.ToLower(text) } }() return out } func (p *TextPipeline) splitWords(in <-chan string) <-chan string { out := make(chan string) go func() { defer close(out) for text := range in { for _, word := range strings.Fields(text) { out <- word } } }() return out } func (p *TextPipeline) removeStopwords(in <-chan string) <-chan string { stopwords := map[string]bool{ "the": true, "a": true, "an": true, "is": true, "are": true, "was": true, "and": true, "or": true, "but": true, } out := make(chan string) go func() { defer close(out) for word := range in { if !stopwords[word] { out <- word } } }() return out } func main() { // 数字 Pipeline fmt.Println("=== 数字 Pipeline ===") pipeline := filterEven(double(square(generate(1, 2, 3, 4, 5)))) sink(pipeline) // 文本 Pipeline fmt.Println("\n=== 文本 Pipeline ===") tp := NewTextPipeline() results := tp.Start() go func() { tp.Feed("The quick brown fox jumps over the lazy dog") tp.Feed("A journey of a thousand miles begins with a single step") tp.Close() }() for word := range results { fmt.Printf("Word: %s\n", word) } }4.2 并发 Pipeline（每个阶段多 worker）
// pipeline/concurrent_pipeline.go package main import ( "fmt" "sync" "time" ) // 并发阶段的 Pipeline type Stage func(<-chan int) <-chan int func concurrentStage(workers int, process func(int) int) Stage { return func(in <-chan int) <-chan int { out := make(chan int) var wg sync.WaitGroup for i := 0; i < workers; i++ { wg.Add(1) go func() { defer wg.Done() for n := range in { out <- process(n) } }() } go func() { wg.Wait() close(out) }() return out } } func main() { // 构建并发 Pipeline // 生成 → 平方(3 workers) → 加倍(2 workers) → 消费 generate := func(nums ...int) <-chan int { out := make(chan int) go func() { defer close(out) for _, n := range nums { out <- n time.Sleep(50 * time.Millisecond) } }() return out } squareStage := concurrentStage(3, func(n int) int { time.Sleep(100 * time.Millisecond) return n * n }) doubleStage := concurrentStage(2, func(n int) int { time.Sleep(50 * time.Millisecond) return n * 2 }) // 组装 Pipeline input := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10) squared := squareStage(input) doubled := doubleStage(squared) // 消费 for result := range doubled { fmt.Printf("Result: %d\n", result) } }五、Worker Pool 模式
5.1 通用 Worker Pool
// pool/worker_pool.go package main import ( "context" "fmt" "sync" "time" ) // Job 定义 type Job struct { ID int Payload string } // Result 定义 type Result struct { JobID int Output string Err error } // Worker Pool type WorkerPool struct { numWorkers int jobs chan Job results chan Result ctx context.Context cancel context.CancelFunc wg sync.WaitGroup } func NewWorkerPool(numWorkers int, bufferSize int) *WorkerPool { ctx, cancel := context.WithCancel(context.Background()) return &WorkerPool{ numWorkers: numWorkers, jobs: make(chan Job, bufferSize), results: make(chan Result, bufferSize), ctx: ctx, cancel: cancel, } } func (wp *WorkerPool) Start() { for i := 0; i < wp.numWorkers; i++ { wp.wg.Add(1) go wp.worker(i) } } func (wp *WorkerPool) worker(id int) { defer wp.wg.Done() fmt.Printf("Worker %d started\n", id) for { select { case <-wp.ctx.Done(): fmt.Printf("Worker %d stopped\n", id) return case job, ok := <-wp.jobs: if !ok { fmt.Printf("Worker %d: jobs channel closed\n", id) return } // 处理 job result := wp.processJob(id, job) // 发送结果 select { case wp.results <- result: case <-wp.ctx.Done(): return } } } } func (wp *WorkerPool) processJob(workerID int, job Job) Result { // 模拟处理 time.Sleep(time.Duration(100+job.ID*10) * time.Millisecond) output := fmt.Sprintf("Worker %d processed job %d: %s", workerID, job.ID, job.Payload) return Result{ JobID: job.ID, Output: output, } } func (wp *WorkerPool) Submit(job Job) { select { case wp.jobs <- job: case <-wp.ctx.Done(): fmt.Printf("Failed to submit job %d: pool stopped\n", job.ID) } } func (wp *WorkerPool) Results() <-chan Result { return wp.results } func (wp *WorkerPool) Stop() { wp.cancel() wp.wg.Wait() close(wp.results) } func (wp *WorkerPool) StopGracefully() { close(wp.jobs) wp.wg.Wait() close(wp.results) } func main() { // 创建 Worker Pool pool := NewWorkerPool(5, 100) pool.Start() // 提交任务 go func() { for i := 1; i <= 20; i++ { pool.Submit(Job{ ID: i, Payload: fmt.Sprintf("task-%d", i), }) } pool.StopGracefully() }() // 收集结果 for result := range pool.Results() { if result.Err != nil { fmt.Printf("Error: %v\n", result.Err) } else { fmt.Printf("Result: %s\n", result.Output) } } }六、超时与取消模式
6.1 Context 的使用
// patterns/context_patterns.go package main import ( "context" "fmt" "time" ) // 1. 超时控制 func operationWithTimeout(ctx context.Context) error { // 创建带超时的 context ctx, cancel := context.WithTimeout(ctx, 2*time.Second) defer cancel() resultCh := make(chan string) errCh := make(chan error) // 执行耗时操作 go func() { result, err := doSlowOperation() if err != nil { errCh <- err return } resultCh <- result }() select { case result := <-resultCh: fmt.Printf("Success: %s\n", result) return nil case err := <-errCh: return fmt.Errorf("operation failed: %w", err) case <-ctx.Done(): return fmt.Errorf("operation timed out: %w", ctx.Err()) } } // 2. 取消传播 func propagateCancellation(ctx context.Context) { // 父 context 取消时，所有子 context 也取消 ctx, cancel := context.WithCancel(ctx) defer cancel() go func() { // 模拟某些条件触发取消 time.Sleep(500 * time.Millisecond) cancel() }() select { case <-ctx.Done(): fmt.Printf("Cancelled: %v\n", ctx.Err()) } } // 3. 携带值 func contextWithValues() { ctx := context.WithValue(context.Background(), "userID", 12345) ctx = context.WithValue(ctx, "requestID", "req-001") processRequest(ctx) } func processRequest(ctx context.Context) { userID := ctx.Value("userID").(int) requestID := ctx.Value("requestID").(string) fmt.Printf("Processing request %s for user %d\n", requestID, userID) } func doSlowOperation() (string, error) { time.Sleep(3 * time.Second) return "done", nil } // 4. 优雅关闭 type Server struct { ctx context.Context cancel context.CancelFunc } func NewServer() *Server { ctx, cancel := context.WithCancel(context.Background()) return &Server{ctx: ctx, cancel: cancel} } func (s *Server) Start() { go func() { for { select { case <-s.ctx.Done(): fmt.Println("Server shutting down...") return default: // 处理请求 time.Sleep(100 * time.Millisecond) } } }() } func (s *Server) Shutdown() { s.cancel() }七、高级并发模式
7.1 Or-Done 模式
// patterns/or_done.go package main import ( "fmt" "time" ) // Or-Done 模式：合并多个 done channel func orDone(channels ...<-chan struct{}) <-chan struct{} { switch len(channels) { case 0: return nil case 1: return channels[0] } done := make(chan struct{}) go func() { defer close(done) switch len(channels) { case 2: select { case <-channels[0]: case <-channels[1]: } default: select { case <-channels[0]: case <-channels[1]: case <-channels[2]: case <-orDone(append(channels[3:], done)...): } } }() return done } // 使用示例 func orDoneExample() { sig := func(after time.Duration) <-chan struct{} { c := make(chan struct{}) go func() { defer close(c) time.Sleep(after) }() return c } start := time.Now() <-orDone( sig(2*time.Hour), sig(5*time.Minute), sig(1*time.Second), sig(1*time.Hour), sig(1*time.Minute), ) fmt.Printf("Done after %v\n", time.Since(start)) }7.2 Bridge 模式
// patterns/bridge.go package main import ( "fmt" ) // Bridge 模式：将 channel 的 channel 扁平化 func bridge(done <-chan struct{}, chanStream <-chan <-chan int) <-chan int { valueStream := make(chan int) go func() { defer close(valueStream) for { var stream <-chan int select { case maybeStream, ok := <-chanStream: if !ok { return } stream = maybeStream case <-done: return } // 从当前 channel 读取值 for val := range orDone(done, stream) { select { case valueStream <- val: case <-done: return } } } }() return valueStream } func orDone(done <-chan struct{}, c <-chan int) <-chan int { valStream := make(chan int) go func() { defer close(valStream) for { select { case <-done: return case v, ok := <-c: if !ok { return } select { case valStream <- v: case <-done: } } } }() return valStream } func main() { // 创建 channel 的 channel genVals := func() <-chan <-chan int { chanStream := make(chan (<-chan int)) go func() { defer close(chanStream) for i := 0; i < 5; i++ { stream := make(chan int, 1) stream <- i close(stream) chanStream <- stream } }() return chanStream } done := make(chan struct{}) defer close(done) for v := range bridge(done, genVals()) { fmt.Printf("Bridge value: %d\n", v) } }7.3 Tee 模式
// patterns/tee.go package main import ( "fmt" ) // Tee 模式：将一个 channel 拆分成两个 func tee(done <-chan struct{}, in <-chan int) (_, _ <-chan int) { out1 := make(chan int) out2 := make(chan int) go func() { defer close(out1) defer close(out2) for val := range orDone(done, in) { // 创建两个副本 var out1, out2 = out1, out2 for i := 0; i < 2; i++ { select { case <-done: return case out1 <- val: out1 = nil // 发送后置 nil，避免重复发送 case out2 <- val: out2 = nil } } } }() return out1, out2 } func main() { done := make(chan struct{}) defer close(done) in := make(chan int) go func() { defer close(in) for i := 1; i <= 5; i++ { in <- i } }() out1, out2 := tee(done, in) // 同时消费两个输出 for i := 0; i < 5; i++ { v1 := <-out1 v2 := <-out2 fmt.Printf("out1: %d, out2: %d\n", v1, v2) } }八、性能基准测试
// benchmarks/concurrency_bench_test.go package benchmarks import ( "sync" "testing" ) func BenchmarkFanIn(b *testing.B) { for i := 0; i < b.N; i++ { ch1 := make(chan int, 100) ch2 := make(chan int, 100) go func() { for j := 0; j < 100; j++ { ch1 <- j } close(ch1) }() go func() { for j := 0; j < 100; j++ { ch2 <- j } close(ch2) }() merged := fanIn(ch1, ch2) for range merged { } } } func BenchmarkWorkerPool(b *testing.B) { for i := 0; i < b.N; i++ { pool := NewWorkerPool(10, 100) pool.Start() for j := 0; j < 100; j++ { pool.Submit(Job{ID: j, Payload: "test"}) } pool.StopGracefully() for range pool.Results() { } } } func BenchmarkPipeline(b *testing.B) { for i := 0; i < b.N; i++ { input := generate(1, 2, 3, 4, 5) squared := square(input) doubled := double(squared) for range doubled { } } } func BenchmarkMutexVsChannel(b *testing.B) { b.Run("Mutex", func(b *testing.B) { var counter int var mu sync.Mutex var wg sync.WaitGroup b.ResetTimer() for i := 0; i < b.N; i++ { wg.Add(1) go func() { mu.Lock() counter++ mu.Unlock() wg.Done() }() } wg.Wait() }) b.Run("Channel", func(b *testing.B) { ch := make(chan int, b.N) var wg sync.WaitGroup b.ResetTimer() for i := 0; i < b.N; i++ { wg.Add(1) go func() { ch <- 1 wg.Done() }() } go func() { wg.Wait() close(ch) }() for range ch { } }) }九、总结
9.1 核心要点
模式 | 描述 | 适用场景 | 面试高频 |
|---|---|---|---|
Fan-In | 多路合并 | 聚合结果 | ⭐⭐⭐⭐⭐ |
Fan-Out | 任务分发 | 并行处理 | ⭐⭐⭐⭐⭐ |
Pipeline | 阶段处理 | 数据流 | ⭐⭐⭐⭐⭐ |
Worker Pool | 资源控制 | 限流削峰 | ⭐⭐⭐⭐⭐ |
Or-Done | 多路取消 | 超时控制 | ⭐⭐⭐ |
Bridge | 扁平化 | 嵌套channel | ⭐⭐⭐ |
Tee | 拆分 | 广播分发 | ⭐⭐⭐ |
9.2 选择指南
如何选择合适的并发模式？ ┌─────────────────────────────────────────────────────────────┐ │ 需要聚合多个结果？ → Fan-In │ │ 需要分发任务？ → Fan-Out + Worker Pool │ │ 需要分阶段处理？ → Pipeline │ │ 需要控制并发数量？ → Worker Pool │ │ 需要超时控制？ → Context + select │ │ 需要优雅关闭？ → Context + Done channel │ │ 需要广播消息？ → Tee 模式 │ │ 需要多路复用？ → Fan-In │ └─────────────────────────────────────────────────────────────┘9.3 记忆口诀
Fan-In 多路合一，聚合结果最给力 Fan-Out 任务分发，并行处理效率高 Pipeline 流水线，阶段处理环环扣 Worker Pool 控资源，限流削峰值稳定 Context 管取消，超时传播两不误 Or-Done 多路停，谁先完成听谁的 Bridge 桥接扁平化，嵌套通道变直线 Tee 拆分广播，一份数据多处收9.4 下讲预告
第10讲：Go 微服务架构 —— 从单体到微服务的演进之路
我们将深入学习：
微服务架构的核心原则
服务拆分策略
服务间通信（REST/gRPC/消息队列）
服务发现与负载均衡
分布式事务
链路追踪
准备好了吗？让我们在第10讲再见！
🧰开发之余的小工具推荐
处理 Base64、JWT 解析、JSON 格式化、Crontab 计算、PDF 合并压缩这些碎片需求，我常用一个纯前端本地工具箱：zz365.top。所有计算在浏览器完成，文件不上服务器，关页即清。免费、无登录、无广告，适合开发者当常驻标签页。
)
