一、并发设计模式概览1.1 为什么需要并发模式并发编程的三大挑战 ┌─────────────────────────────────────────────────────────────┐ │ 1. 竞态条件 (Race Condition) │ │ 多个 goroutine 同时读写共享数据 │ │ 解决方案: mutex / channel / atomic │ │ │ │ 2. 死锁 (Deadlock) │ │ goroutine 互相等待对方释放资源 │ │ 解决方案: 按序加锁 / timeout / 锁粒度控制 │ │ │ │ 3. 资源泄漏 (Resource Leak) │ │ goroutine 永远无法退出 │ │ 解决方案: context 取消 / 超时控制 / 优雅关闭 │ └─────────────────────────────────────────────────────────────┘ Go 并发哲学: Dont 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(100job*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, retry1) 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(100job.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。所有计算在浏览器完成文件不上服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。 第9讲：Go 并发设计模式 —— 从 Fan-in/Fan-out 到 Pipeline 的生产级实践
一、并发设计模式概览1.1 为什么需要并发模式并发编程的三大挑战 ┌─────────────────────────────────────────────────────────────┐ │ 1. 竞态条件 (Race Condition) │ │ 多个 goroutine 同时读写共享数据 │ │ 解决方案: mutex / channel / atomic │ │ │ │ 2. 死锁 (Deadlock) │ │ goroutine 互相等待对方释放资源 │ │ 解决方案: 按序加锁 / timeout / 锁粒度控制 │ │ │ │ 3. 资源泄漏 (Resource Leak) │ │ goroutine 永远无法退出 │ │ 解决方案: context 取消 / 超时控制 / 优雅关闭 │ └─────────────────────────────────────────────────────────────┘ Go 并发哲学: Dont 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(100job*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, retry1) 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(100job.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。所有计算在浏览器完成文件不上服务器关页即清。免费、无登录、无广告适合开发者当常驻标签页。