Go教程(三)并发编程

Go 03：并发编程

并发是 Go 最大的特色。对比 Python：

• Python 有 GIL，CPU 密集型任务需多进程；asyncio 需要整条调用链都是 async。
• Go 的 goroutine 是轻量协程（初始约 2KB 栈），可轻松开几万个，自动利用多核，普通同步代码直接 go 一下就并发了，没有"函数染色"问题。

口诀：不要通过共享内存来通信，而要通过通信来共享内存。

1. goroutine 与 WaitGroup

func worker(id int) {
	time.Sleep(100 * time.Millisecond)
	fmt.Println("worker", id, "done")
}

func main() {
	var wg sync.WaitGroup
	for i := 1; i <= 5; i++ {
		wg.Add(1)
		go func() {          // Go 1.22+ 循环变量每次迭代独立，可直接使用 i
			defer wg.Done()
			worker(i)
		}()
	}
	wg.Wait()                // ≈ asyncio.gather / thread.join
	fmt.Println("all done")
}

Go 1.25+ 可用 wg.Go(func(){ ... }) 简化 Add/Done。

2. channel：goroutine 之间的管道

ch := make(chan int)        // 无缓冲：发送和接收必须同时就绪
buf := make(chan int, 10)   // 有缓冲：≈ queue.Queue(maxsize=10)

go func() {
	for i := 0; i < 3; i++ {
		ch <- i             // 发送
	}
	close(ch)               // 发送方负责关闭
}()

for v := range ch {         // 接收直到 channel 关闭
	fmt.Println(v)
}

3. select：多路复用 + 超时

select {
case msg := <-msgCh:
	fmt.Println("got", msg)
case err := <-errCh:
	fmt.Println("error", err)
case <-time.After(3 * time.Second):
	fmt.Println("timeout")
}

4. 互斥锁：保护共享数据

type SafeCounter struct {
	mu sync.Mutex
	m  map[string]int
}

func (c *SafeCounter) Inc(key string) {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.m[key]++
}

检测数据竞争：go run -race main.go / go test -race ./...（强烈推荐养成习惯）。

5. context：取消与超时（服务端必备）

context.Context 用于在调用链中传递取消信号、截止时间、请求范围的值。HTTP 请求、数据库查询、RPC 调用几乎都接收 ctx 作为第一个参数。

func slowQuery(ctx context.Context) (string, error) {
	select {
	case <-time.After(2 * time.Second):
		return "result", nil
	case <-ctx.Done():
		return "", ctx.Err()    // context.DeadlineExceeded 或 context.Canceled
	}
}

ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
_, err := slowQuery(ctx)
fmt.Println(err)    // context deadline exceeded

6. 常用并发模式

6.1 Worker Pool（限制并发数）—— 最常用

应用场景：批量调用第三方 API、批量处理文件，避免同时打开过多连接。

type Job struct{ URL string }
type Result struct {
	URL    string
	Status int
	Err    error
	Cost   time.Duration
}

func check(client *http.Client, job Job) Result {
	start := time.Now()
	resp, err := client.Get(job.URL)
	if err != nil {
		return Result{URL: job.URL, Err: err, Cost: time.Since(start)}
	}
	defer resp.Body.Close()
	return Result{URL: job.URL, Status: resp.StatusCode, Cost: time.Since(start)}
}

func RunPool(urls []string, workers int) []Result {
	jobs := make(chan Job)
	results := make(chan Result)
	client := &http.Client{Timeout: 5 * time.Second}

	var wg sync.WaitGroup
	for range workers {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for job := range jobs {
				results <- check(client, job)
			}
		}()
	}

	go func() {
		for _, u := range urls {
			jobs <- Job{URL: u}
		}
		close(jobs)
	}()

	go func() {
		wg.Wait()
		close(results)       // 所有 worker 结束后关闭结果通道
	}()

	var all []Result
	for r := range results {
		all = append(all, r)
	}
	return all
}

6.2 errgroup：并发执行，任一失败即取消其他

go get golang.org/x/sync/errgroup

// 场景：渲染博客首页需要并发拉取 文章列表、热门标签、用户信息
func LoadHomePage(ctx context.Context) (*HomePage, error) {
	g, ctx := errgroup.WithContext(ctx)
	var page HomePage

	g.Go(func() error {
		posts, err := fetchPosts(ctx)
		page.Posts = posts
		return err
	})
	g.Go(func() error {
		tags, err := fetchTags(ctx)
		page.Tags = tags
		return err
	})
	g.Go(func() error {
		user, err := fetchUser(ctx)
		page.User = user
		return err
	})

	if err := g.Wait(); err != nil {   // 返回第一个错误
		return nil, err
	}
	return &page, nil
}

g.SetLimit(n) 还可以直接限制并发数，很多场景可以替代手写 worker pool。

6.3 Pipeline（流水线）

// 读取 -> 解析 -> 过滤 -> 输出，每个阶段一个 goroutine
func generate(lines []string) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for _, l := range lines {
			out <- l
		}
	}()
	return out
}

func filterErrors(in <-chan string) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for l := range in {
			if strings.Contains(l, "ERROR") {
				out <- l
			}
		}
	}()
	return out
}

for line := range filterErrors(generate(lines)) {
	fmt.Println(line)
}

6.4 定时任务 + 优雅退出

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			fmt.Println("clean expired sessions...")
		case <-ctx.Done():
			fmt.Println("shutting down")
			return
		}
	}
}

6.5 sync.Once：单例 / 懒加载

var (
	dbOnce sync.Once
	db     *sql.DB
)

func GetDB() *sql.DB {
	dbOnce.Do(func() {
		db, _ = sql.Open("sqlite", "app.db")
	})
	return db
}

7. 常见坑

8. 练习

1. 用 worker pool 并发下载一批图片，限制并发 5，统计成功/失败数。
2. 用 errgroup + SetLimit 重写上面的 URL 检查。
3. 实现一个令牌桶限流器：每秒最多允许 N 次调用（提示：time.Ticker + 带缓冲 channel，或使用 golang.org/x/time/rate）。