Go语言WaitGroup使用指南实现并发任务同步

首页

编程语言

热心网友

转载

2026-05-10

在Go语言的并发世界里，协调多个goroutine的执行顺序是个绕不开的话题。你可能会遇到这样的场景：主程序需要等待一批后台任务全部完成，才能继续下一步操作。这时候，如果还用传统的睡眠或者忙等，代码就显得笨拙且低效。好在Go标准库提供了一个优雅的解决方案——sync.WaitGroup。这个看似简单的同步原语，却是构建健壮并发程序的基石之一。

免费影视、动漫、音乐、游戏、小说资源长期稳定更新！ 👉 点此立即查看 👈

Go语言中WaitGroup并发同步利器

今天，我们就来深入聊聊WaitGroup，从它的基本用法到底层原理，再到实战中的最佳实践和常见“坑点”，帮你彻底掌握这个并发利器。

1. WaitGroup的基本概念

简单来说，sync.WaitGroup就是一个“任务完成等待器”。它内部维护一个计数器，用来追踪尚未完成的goroutine数量。主goroutine通过调用Wait()方法进入阻塞状态，直到计数器归零，意味着所有子任务都已执行完毕，程序才会继续向下执行。这种机制完美替代了低效的轮询检查，让并发同步变得清晰而直接。

2. WaitGroup的基本用法

2.1 创建和使用WaitGroup

先来看一个最经典的例子，感受一下它的工作流程：

package main

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(1 * time.Second)
			fmt.Printf("Goroutine %d finished\n", id)
		}(i)
	}
	
	fmt.Println("Waiting for all goroutines to finish...")
	wg.Wait() // 等待所有协程完成
	fmt.Println("All goroutines finished")
}

运行这段代码，你会看到主程序在打印出等待信息后，会耐心地等所有5个goroutine都执行完（包括那1秒的睡眠），最后才打印完成信息。整个过程井然有序。

2.2 WaitGroup的方法

WaitGroup的API非常简洁，只有三个方法：

Add(delta int)：为计数器增加一个值（通常是正数）。
Done()：将计数器减1，它实际上就是Add(-1)的便捷封装。
Wait()：阻塞当前goroutine，直到计数器清零。

2.3 WaitGroup的特性

理解这几个特性，能帮你避免很多低级错误：

计数器是核心：所有操作都围绕这个计数器展开。
阻塞是常态：Wait()会一直阻塞，这是它的本职工作。
一次性用品：这一点至关重要——一旦计数器归零且Wait()返回，这个WaitGroup就不能再被使用了。如果需要等待另一组任务，请创建一个新的。

3. WaitGroup的原理

3.1 WaitGroup的底层实现

别看WaitGroup用起来简单，它的内部实现可是标准的同步模式。它主要包含三部分：

一个计数器，记录活跃的goroutine数。
一把互斥锁（sync.Mutex），保护计数器在并发读写时的安全。
一个条件变量（sync.Cond），用于在计数器变化时通知等待的goroutine。

3.2 WaitGroup的工作原理

了解了结构，它的工作流程就很好理解了：

Add操作：
- 加锁，保证计数器操作的原子性。
- 增加计数器值。
- 解锁。
Done操作：
- 加锁。
- 将计数器减1。
- 如果发现计数器减到了0，就通过条件变量广播（Broadcast），唤醒所有正在Wait()的goroutine。
- 解锁。
Wait操作：
- 加锁。
- 检查计数器，如果大于0，就调用条件变量的Wait()方法释放锁并进入休眠。
- 被唤醒后，解锁并返回。

4. WaitGroup的高级用法

掌握了基础，我们来看看在实际项目中，如何让WaitGroup与其他Go并发组件配合，解决更复杂的问题。

4.1 WaitGroup与错误处理

直接使用WaitGroup，子goroutine的错误是无法直接返回给主goroutine的。一个常见的模式是结合带缓冲的通道（channel）来收集错误：

package main

import (
	"fmt"
	"sync"
	"time"
)

func main() {
	var wg sync.WaitGroup
	errCh := make(chan error, 5) // 缓冲通道，防止goroutine阻塞
	
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			time.Sleep(1 * time.Second)
			
			// 模拟错误
			if id == 2 {
				errCh <- fmt.Errorf("error in goroutine %d", id)
				return
			}
			fmt.Printf("Goroutine %d finished\n", id)
		}(i)
	}
	
	// 专门用一个goroutine等待，然后关闭错误通道
	go func() {
		wg.Wait()
		close(errCh)
	}()
	
	// 主goroutine可以安全地遍历已关闭的通道，收集所有错误
	for err := range errCh {
		fmt.Println("Error:", err)
	}
	fmt.Println("All goroutines finished")
}

4.2 WaitGroup与Context

在需要超时或取消控制的场景，context.Context是WaitGroup的好搭档：

package main

import (
	"context"
	"fmt"
	"sync"
	"time"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel() // 确保资源释放
	
	var wg sync.WaitGroup
	
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			
			// 监听上下文取消和任务完成两个事件
			select {
			case <-ctx.Done(): // 超时或被取消
				fmt.Printf("Goroutine %d cancelled\n", id)
				return
			case <-time.After(2 * time.Second): // 模拟任务耗时
				fmt.Printf("Goroutine %d finished\n", id)
			}
		}(i)
	}
	
	wg.Wait() // 这里会等待所有goroutine结束（无论是否超时）
	fmt.Println("All goroutines finished")
}

4.3 WaitGroup与通道

除了传递错误，通道更常用于收集并发任务的结果：

package main

import (
	"fmt"
	"sync"
	"time"
)

func main() {
	var wg sync.WaitGroup
	resultCh := make(chan int, 5) // 缓冲通道，容量等于任务数
	
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			time.Sleep(1 * time.Second)
			resultCh <- id * 2 // 发送结果
		}(i)
	}
	
	// 同样，用另一个goroutine等待并关闭结果通道
	go func() {
		wg.Wait()
		close(resultCh)
	}()
	
	// 主goroutine遍历通道，获取所有结果
	for result := range resultCh {
		fmt.Println("Result:", result)
	}
	fmt.Println("All goroutines finished")
}

5. WaitGroup的最佳实践

用好WaitGroup，关键在于遵循一些约定俗成的规则，这能帮你避开绝大多数陷阱。

5.1 正确使用Add方法

先Add，后启动：务必在启动goroutine之前调用Add。如果顺序反了，可能出现goroutine已经调用Done使计数器归零，而主goroutine还没调用Add，导致Wait无法阻塞。
使用正数：Add的参数应该是你计划等待的goroutine数量，通常为正数。
别用负数：减少计数请用Done()，不要直接Add(-1)，虽然语法允许，但会降低代码可读性。

5.2 正确使用Done方法

确保执行：每个goroutine在退出前，无论正常还是异常，都必须调用Done。
defer是黄金搭档：使用defer wg.Done()是保证这一点的最佳实践。即使goroutine中途panic，defer语句也会执行。

5.3 正确使用Wait方法

最后调用：在所有goroutine都启动之后，再调用Wait。
一次性：牢记WaitGroup不可重用。一次Wait结束后，它的使命就完成了。

5.4 错误处理

通道收集：如前所述，使用带缓冲的通道来收集子goroutine的错误或结果，是标准模式。
Context超时：对于可能长时间运行或阻塞的任务，务必结合Context设置超时，防止程序永远卡住。

5.5 性能优化

控制goroutine数量：goroutine虽轻量，但并非无限。对于海量小任务，考虑使用worker pool（协程池）模式。
避免长时阻塞：确保goroutine内的任务不会无限期阻塞，否则会拖累整个等待组。

6. WaitGroup的常见问题与解决方案

即便知道了最佳实践，实际编码中还是可能遇到一些问题。下面这几个场景非常典型。

6.1 计数器不匹配

问题：Add和Done的调用次数没对上。比如Add(3)却只Done()了两次，计数器永远不为零，Wait()就会死锁。

解决方案：

仔细核对循环次数和Add的调用。
坚持使用defer wg.Done()。

6.2 重复使用WaitGroup

问题：在一个WaitGroup的Wait()返回后，又用它去等待另一组任务。这是未定义行为，可能导致程序崩溃或死锁。

解决方案：

为每一组独立的并发任务创建新的WaitGroup实例。
把WaitGroup当作一次性消耗品来对待。

6.3 协程泄露

问题：goroutine因为某种原因（如死锁、无限循环、等待一个永远不会关闭的channel）而无法退出，导致Done永远不被调用，主程序永久阻塞在Wait。

解决方案：

为goroutine内的阻塞操作（如网络请求、通道操作）设置超时（使用context.Context或select+time.After）。
进行代码审查，确保goroutine都有明确的退出路径。

6.4 性能问题

问题：盲目启动成千上万个goroutine去处理IO密集型或CPU密集型任务，导致系统调度开销巨大，甚至资源耗尽。

解决方案：

对于CPU密集型任务，goroutine数量最好接近CPU核心数。
对于IO密集型任务，可以使用数量稍多的goroutine，但也要有上限。
考虑使用“生产者-消费者”模型和固定大小的goroutine池。

7. WaitGroup的实战应用

理论说再多，不如看几个实际例子。

7.1 并行处理任务

这是最直接的场景，比如批量处理一批数据：

package main

import (
	"fmt"
	"sync"
	"time"
)

func processTask(id int) {
	fmt.Printf("Processing task %d\n", id)
	time.Sleep(1 * time.Second) // 模拟处理耗时
	fmt.Printf("Task %d processed\n", id)
}

func main() {
	var wg sync.WaitGroup
	tasks := []int{1, 2, 3, 4, 5}
	
	for _, task := range tasks {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			processTask(id)
		}(task)
	}
	
	fmt.Println("Waiting for all tasks to complete...")
	wg.Wait()
	fmt.Println("All tasks completed")
}

7.2 并发下载文件

利用并发加速IO密集型操作，比如下载多个文件：

package main

import (
	"fmt"
	"sync"
	"time"
)

func downloadFile(url string) {
	fmt.Printf("Downloading %s\n", url)
	time.Sleep(2 * time.Second) // 模拟下载时间
	fmt.Printf("Downloaded %s\n", url)
}

func main() {
	var wg sync.WaitGroup
	urls := []string{
		"https://example.com/file1.txt",
		"https://example.com/file2.txt",
		// ... 更多URL
	}
	
	for _, url := range urls {
		wg.Add(1)
		go func(u string) {
			defer wg.Done()
			downloadFile(u)
		}(url)
	}
	
	fmt.Println("Waiting for all downloads to complete...")
	wg.Wait()
	fmt.Println("All downloads completed")
}

7.3 并发数据库操作

并发执行多个独立的数据库查询，显著提升聚合查询效率：

package main

import (
	"fmt"
	"sync"
	"time"
)

func queryDatabase(id int) {
	fmt.Printf("Querying database for id %d\n", id)
	time.Sleep(500 * time.Millisecond) // 模拟查询延迟
	fmt.Printf("Query completed for id %d\n", id)
}

func main() {
	var wg sync.WaitGroup
	ids := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
	
	for _, id := range ids {
		wg.Add(1)
		go func(i int) {
			defer wg.Done()
			queryDatabase(i)
		}(id)
	}
	
	fmt.Println("Waiting for all database queries to complete...")
	wg.Wait()
	fmt.Println("All database queries completed")
}