阶段 1 并发地基 · 代码实例集

覆盖：goroutine / channel / select / GMP / sync / context / 并发模式 / race 检测。 配合 02-语言特性/02-并发模型.md、03-标准库/04-并发原语.md 食用。 所有代码均可直接 go run 运行。重点看注释。

---

1. goroutine 基础

1.1 启动 goroutine + WaitGroup 等待

package main

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

func main() {
	var wg sync.WaitGroup
	for i := 0; i < 3; i++ {
		wg.Add(1)
		go func(id int) { // 参数传入，避免闭包陷阱
			defer wg.Done()
			time.Sleep(100 * time.Millisecond)
			fmt.Println("worker", id, "done")
		}(i)
	}
	wg.Wait() // 阻塞直到所有 Done()
	fmt.Println("all done")
}

1.2 闭包变量捕获陷阱（经典坑）

// ❌ 错误：所有 goroutine 共享同一个 i，循环结束 i=3，打印全是 3
for i := 0; i < 3; i++ {
	go func() { fmt.Println(i) }() // 捕获的是 i 的引用
}

// ✅ 正确：把 i 作为参数传进去（值拷贝）
for i := 0; i < 3; i++ {
	go func(n int) { fmt.Println(n) }(i)
}

1.3 goroutine 泄露（忘记退出）

// ❌ ch 没人关闭、没人接收，goroutine 永远阻塞 → 泄露
func leak() {
	ch := make(chan int)
	go func() {
		val := <-ch      // 永远等不到
		fmt.Println(val) // 永不执行
	}()
	// 函数返回，goroutine 仍挂着
}

排查：go tool pprof 看 goroutine 数量，或用 runtime.NumGoroutine() 打点观察。

---

2. channel

2.1 无缓冲 channel 同步（发送阻塞到接收者就绪）

func main() {
	ch := make(chan int) // 无缓冲
	go func() {
		ch <- 42 // 阻塞，直到下面有人接收
	}()
	fmt.Println(<-ch) // 42
}

2.2 有缓冲 channel

ch := make(chan int, 2)
ch <- 1 // 不阻塞（缓冲区有空位）
ch <- 2 // 不阻塞
// ch <- 3 // 阻塞：缓冲满了
fmt.Println(<-ch, <-ch) // 1 2（FIFO）

2.3 关闭语义 + for-range 接收（最安全）

ch := make(chan int, 3)
ch <- 1
ch <- 2
close(ch)

// for-range 在 channel 关闭后自动退出
for v := range ch {
	fmt.Println(v) // 1 2
}
// 关闭后再接收：返回零值，ok=false（不 panic）
v, ok := <-ch
fmt.Println(v, ok) // 0 false

操作	结果
向已关闭 channel 发送	panic: send on closed channel
关闭已关闭 channel	panic: close of closed channel
从已关闭 channel 接收	立即返回零值，ok=false
向 nil channel 收发	永久阻塞

2.4 单向 channel（限制方向，接口设计用）

func producer(out chan<- int) { // 只发
	out <- 1
	close(out)
}
func consumer(in <-chan int) { // 只收
	for v := range in {
		fmt.Println(v)
	}
}
func main() {
	ch := make(chan int, 1)
	producer(ch)
	consumer(ch)
}

2.5 select + default（非阻塞）

ch := make(chan int, 1)
select {
case v := <-ch:
	fmt.Println("received", v)
default:
	fmt.Println("no data, non-blocking") // ch 空时走这里
}

2.6 select + time.After（超时控制）

ch := make(chan string, 1)
go func() { time.Sleep(200 * time.Millisecond); ch <- "result" }()

select {
case res := <-ch:
	fmt.Println(res)
case <-time.After(100 * time.Millisecond):
	fmt.Println("timeout") // 100ms < 200ms → 走超时
}

---

3. GMP 简述（概念 + 观测）

Go 调度模型三层：

• G (goroutine)：用户态轻量协程，初始栈 ~2KB，可动态扩缩。
• M (machine)：操作系统线程，真正跑代码的实体。
• P (processor)：逻辑处理器，持有可运行 G 的本地队列，M 必须绑定 P 才能执行 G。

import (
	"fmt"
	"runtime"
)

func main() {
	fmt.Println("NumCPU:", runtime.NumCPU())
	fmt.Println("GOMAXPROCS:", runtime.GOMAXPROCS(0))
	runtime.GOMAXPROCS(runtime.NumCPU()) // 通常默认已是 CPU 数
	fmt.Println("goroutines:", runtime.NumGoroutine())
}

关键点：goroutine 数量不是无成本，海量 goroutine 会吃内存；GOMAXPROCS 控制并行度。

---

4. sync 原语

4.1 Mutex 保护共享计数

var (
	mu  sync.Mutex
	cnt int
)

func worker() {
	for i := 0; i < 1000; i++ {
		mu.Lock()
		cnt++ // 临界区
		mu.Unlock()
	}
}

不加锁跑 go test -race 必报 DATA RACE。

4.2 RWMutex 读写锁

var (
	mu   sync.RWMutex
	cache = map[string]string{}
)

func read(k string) string {
	mu.RLock()         // 多读可并发
	defer mu.RUnlock()
	return cache[k]
}
func write(k, v string) {
	mu.Lock()          // 写独占
	defer mu.Unlock()
	cache[k] = v
}

4.3 WaitGroup 等待多任务（见 1.1）

4.4 Once 单次初始化

var once sync.Once
var config map[string]string

func getConfig() map[string]string {
	once.Do(func() { // 无论多少 goroutine 调用，只执行一次
		config = loadConfig()
	})
	return config
}

4.5 sync.Pool 对象复用（降低 GC 压力）

var bufPool = sync.Pool{
	New: func() interface{} { return new(bytes.Buffer) },
}

func process() {
	buf := bufPool.Get().(*bytes.Buffer)
	buf.Reset()
	buf.WriteString("hello")
	// ... 使用 buf
	bufPool.Put(buf) // 用完归还
}

4.6 atomic 原子操作（无锁计数器）

import "sync/atomic"

var counter int64
atomic.AddInt64(&counter, 1)        // 原子加
v := atomic.LoadInt64(&counter)     // 原子读
atomic.StoreInt64(&counter, 0)      // 原子写

高频计数场景用 atomic 比 Mutex 性能更好。

---

5. context

5.1 WithCancel 取消传播

ctx, cancel := context.WithCancel(context.Background())
defer cancel() // 必须 defer，否则 timer/资源泄漏

go func() {
	for {
		select {
		case <-ctx.Done(): // 取消信号
			fmt.Println("canceled:", ctx.Err())
			return
		default:
			work()
		}
	}
}()
cancel() // 触发 Done 关闭，goroutine 监听到后 return

5.2 WithTimeout 超时

ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
defer cancel()

select {
case res := <-doWork(ctx):
	fmt.Println(res)
case <-ctx.Done():
	fmt.Println("timeout:", ctx.Err()) // context.DeadlineExceeded
}

5.3 WithValue 传值（谨慎用，仅传请求域元数据）

ctx := context.WithValue(context.Background(), "requestID", "abc-123")
// 下游：rid := ctx.Value("requestID").(string)

注意：不要用 context 传可选参数（那是反模式），只传请求生命周期内的元数据（traceID、user 等）。

5.4 取消后必须 return，否则泄露

// ❌ 收到取消信号但没 return，goroutine 继续跑 → 泄露
select {
case <-ctx.Done():
default:
	work()
}

// ✅
select {
case <-ctx.Done():
	return
default:
	work()
}

---

6. 并发模式

6.1 Worker Pool（任务池，限制并发数）

func workerPool(jobs <-chan int, results chan<- int, workers int) {
	var wg sync.WaitGroup
	for w := 0; w < workers; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := range jobs { // jobs 关闭后自动退出
				results <- j * j
			}
		}()
	}
	go func() { wg.Wait(); close(results) }()
}

func main() {
	jobs := make(chan int, 10)
	results := make(chan int, 10)
	workerPool(jobs, results, 3)
	for i := 1; i <= 5; i++ { jobs <- i }
	close(jobs)
	for r := range results { fmt.Println(r) }
}

6.2 扇出 / 扇入（fan-out / fan-in）

// 一个输入，多个 worker 处理（扇出）；多个 worker 结果汇到一个 channel（扇入）
func fanIn(chans ...<-chan int) <-chan int {
	out := make(chan int)
	var wg sync.WaitGroup
	for _, c := range chans {
		wg.Add(1)
		go func(ch <-chan int) {
			defer wg.Done()
			for v := range ch { out <- v }
		}(c)
	}
	go func() { wg.Wait(); close(out) }()
	return out
}

6.3 并发任务池 + recover 隔离 + cancel 传播（综合）

// 见 教学讲义/阶段1-第1课-RunTasks骨架.md，填空完成。
// 要点：recover 转 panic 为 error + cancel() 通知其余 + Mutex 保护 errs。

---

7. 数据竞争检测

7.1 用 race detector

go test -race ./...      # 测试时检测
go run -race main.go     # 运行时检测

报告格式：WARNING: DATA RACE + 读写 goroutine 栈。

7.2 用 channel 替代共享内存（Go 哲学）

“Share memory by communicating, don’t communicate by sharing memory.” 能用 channel 传递所有权，就少用 mutex 共享变量，race 从根上消失。

---

自测题

1. 无缓冲 channel 的发送何时解除阻塞？
2. 向已关闭 channel 发送会怎样？从已关闭 channel 接收呢？
3. recover 为什么必须在 defer 函数体内直接调用？
4. context cancel 后，子 goroutine 会自动退出吗？为什么必须 return？
5. sync.Once 的典型用途是什么？
6. 下面的代码有什么问题？怎么改？

for i := 0; i < 3; i++ {
	go func() { fmt.Println(i) }()
}

答案自查：回看对应小节实例与 03-标准库/04-并发原语.md。卡住就问我。