Go语言中的并发模式：从WaitGroup到errgroup

Go语言中的并发模式从WaitGroup到errgroup作为一个写了十几年代码的Go后端老兵我深刻体会到并发编程的重要性。Go语言以其简洁的并发模型著称通过goroutine和channel我们可以轻松实现高效的并发程序。今天咱们就聊聊Go语言中的并发模式从WaitGroup到errgroup帮助你写出更健壮的并发代码。基本并发原语1. WaitGroupsync.WaitGroup是Go语言中最基础的并发同步原语用于等待一组goroutine完成。import ( fmt sync time ) func main() { var wg sync.WaitGroup for i : 0; i 5; i { wg.Add(1) // 增加计数器 go func(id int) { defer wg.Done() // 减少计数器 fmt.Printf(Goroutine %d started\n, id) time.Sleep(time.Second) fmt.Printf(Goroutine %d finished\n, id) }(i) } wg.Wait() // 等待所有goroutine完成 fmt.Println(All goroutines finished) }2. Mutexsync.Mutex用于保护共享资源防止并发访问导致的数据竞争。import ( fmt sync time ) var ( counter int mutex sync.Mutex ) func main() { var wg sync.WaitGroup for i : 0; i 1000; i { wg.Add(1) go func() { defer wg.Done() mutex.Lock() defer mutex.Unlock() counter }() } wg.Wait() fmt.Println(Counter:, counter) }3. RWMutexsync.RWMutex是读写锁允许多个读操作同时进行但写操作是互斥的。import ( fmt sync time ) var ( data make(map[string]string) rwMutex sync.RWMutex ) func readData(key string) string { rwMutex.RLock() defer rwMutex.RUnlock() return data[key] } func writeData(key, value string) { rwMutex.Lock() defer rwMutex.Unlock() data[key] value } func main() { // 写入数据 writeData(name, Alice) var wg sync.WaitGroup // 多个读操作 for i : 0; i 10; i { wg.Add(1) go func(id int) { defer wg.Done() fmt.Printf(Reader %d: %s\n, id, readData(name)) }(i) } // 一个写操作 wg.Add(1) go func() { defer wg.Done() time.Sleep(time.Millisecond * 100) writeData(name, Bob) fmt.Println(Writer: Updated name to Bob) }() wg.Wait() fmt.Println(Final data:, data[name]) }高级并发模式1. errgrouperrgroup是Go 1.7引入的包用于管理一组goroutine并处理错误。import ( context fmt net/http golang.org/x/sync/errgroup ) func main() { g, ctx : errgroup.WithContext(context.Background()) urls : []string{ https://example.com, https://golang.org, https://invalid-url, } for _, url : range urls { url : url // 捕获循环变量 g.Go(func() error { req, err : http.NewRequest(GET, url, nil) if err ! nil { return err } req req.WithContext(ctx) resp, err : http.DefaultClient.Do(req) if err ! nil { return err } defer resp.Body.Close() fmt.Printf(%s: %d\n, url, resp.StatusCode) return nil }) } if err : g.Wait(); err ! nil { fmt.Printf(Error: %v\n, err) } else { fmt.Println(All requests completed successfully) } }2. Oncesync.Once用于确保某个函数只执行一次。import ( fmt sync ) var ( once sync.Once instance *Database ) type Database struct { // 数据库连接 } func GetDatabase() *Database { once.Do(func() { instance Database{} fmt.Println(Database initialized) }) return instance } func main() { var wg sync.WaitGroup for i : 0; i 10; i { wg.Add(1) go func() { defer wg.Done() db : GetDatabase() fmt.Printf(Database instance: %p\n, db) }() } wg.Wait() }3. Poolsync.Pool用于对象复用减少内存分配和GC压力。import ( fmt sync ) var bufferPool sync.Pool{ New: func() interface{} { fmt.Println(Creating new buffer) return make([]byte, 1024) }, } func main() { var wg sync.WaitGroup for i : 0; i 5; i { wg.Add(1) go func(id int) { defer wg.Done() buf : bufferPool.Get().([]byte) defer bufferPool.Put(buf) fmt.Printf(Goroutine %d: Using buffer %p\n, id, buf) }(i) } wg.Wait() // 再次使用 buf : bufferPool.Get().([]byte) fmt.Printf(Main: Using buffer %p\n, buf) bufferPool.Put(buf) }实用并发模式1. 工作池模式工作池模式用于限制并发数量避免创建过多goroutine。import ( fmt sync time ) func worker(id int, jobs -chan int, results chan- int) { for job : range jobs { fmt.Printf(Worker %d processing job %d\n, id, job) time.Sleep(time.Second) results - job * 2 } } func main() { const ( numWorkers 3 numJobs 10 ) jobs : make(chan int, numJobs) results : make(chan int, numJobs) // 启动工作池 var wg sync.WaitGroup for i : 1; i numWorkers; i { wg.Add(1) go func(id int) { defer wg.Done() worker(id, jobs, results) }(i) } // 发送任务 for i : 1; i numJobs; i { jobs - i } close(jobs) // 收集结果 go func() { wg.Wait() close(results) }() for result : range results { fmt.Printf(Result: %d\n, result) } }2. 扇入扇出模式扇入扇出模式用于处理多个输入合并为一个输出。import ( fmt sync time ) func producer(id int, out chan- int) { for i : 0; i 5; i { value : id*10 i fmt.Printf(Producer %d: %d\n, id, value) out - value time.Sleep(time.Millisecond * 100) } } func fanIn(inputs ...-chan int) -chan int { out : make(chan int) var wg sync.WaitGroup for _, input : range inputs { wg.Add(1) go func(ch -chan int) { defer wg.Done() for value : range ch { out - value } }(input) } go func() { wg.Wait() close(out) }() return out } func main() { ch1 : make(chan int) ch2 : make(chan int) ch3 : make(chan int) // 启动生产者 go producer(1, ch1) go producer(2, ch2) go producer(3, ch3) // 扇入 merged : fanIn(ch1, ch2, ch3) // 消费结果 for value : range merged { fmt.Printf(Consumer: %d\n, value) } }3. 超时控制模式超时控制模式用于防止goroutine无限阻塞。import ( context fmt time ) func main() { ctx, cancel : context.WithTimeout(context.Background(), 2*time.Second) defer cancel() resultCh : make(chan string) go func() { time.Sleep(3 * time.Second) // 模拟长时间操作 resultCh - Operation completed }() select { case result : -resultCh: fmt.Println(Result:, result) case -ctx.Done(): fmt.Println(Operation timed out) } }并发最佳实践限制并发数量使用工作池模式限制并发goroutine数量避免系统资源耗尽。使用context管理生命周期使用context.Context来管理goroutine的生命周期支持取消和超时。避免数据竞争使用Mutex、RWMutex或channel来保护共享资源。处理错误使用errgroup来管理一组goroutine的错误。优雅关闭确保所有goroutine都能正常关闭避免泄漏。使用原子操作对于简单的计数器等操作使用sync/atomic包提高性能。避免死锁注意goroutine之间的依赖关系避免循环等待。测试并发代码使用-race标志运行测试检测数据竞争。实战案例并发下载文件import ( context fmt io net/http os path/filepath golang.org/x/sync/errgroup ) func downloadFile(ctx context.Context, url string, outputDir string) error { req, err : http.NewRequest(GET, url, nil) if err ! nil { return err } req req.WithContext(ctx) resp, err : http.DefaultClient.Do(req) if err ! nil { return err } defer resp.Body.Close() if resp.StatusCode ! http.StatusOK { return fmt.Errorf(bad status: %s, resp.Status) } filename : filepath.Base(url) filepath : filepath.Join(outputDir, filename) file, err : os.Create(filepath) if err ! nil { return err } defer file.Close() _, err io.Copy(file, resp.Body) if err ! nil { return err } fmt.Printf(Downloaded %s to %s\n, url, filepath) return nil } func main() { urls : []string{ https://example.com, https://golang.org, https://github.com, } outputDir : ./downloads if err : os.MkdirAll(outputDir, 0755); err ! nil { fmt.Println(Error creating output directory:, err) return } g, ctx : errgroup.WithContext(context.Background()) for _, url : range urls { url : url // 捕获循环变量 g.Go(func() error { return downloadFile(ctx, url, outputDir) }) } if err : g.Wait(); err ! nil { fmt.Println(Error:, err) } else { fmt.Println(All downloads completed successfully) } }总结并发编程是Go语言的核心特性之一通过掌握各种并发模式我们可以写出更高效、更健壮的并发代码。从基础的WaitGroup到高级的errgroup从工作池模式到扇入扇出模式Go语言提供了丰富的并发工具和模式。记住并发编程需要谨慎处理要注意避免数据竞争、死锁等问题。希望本文能帮助你在Go语言项目中更好地应用并发模式构建更高效的应用程序。