并发编程

掌握 Go 的并发模型，Goroutine 与 Channel

5 课时

Goroutine

Goroutine 是 Go 实现并发的核心。它是一种轻量级线程，由 Go 运行时管理。

启动一个 Goroutine 只需在函数调用前加上 go 关键字。

Goroutine 基础

1package main
2 
3import (
4    "fmt"
5    "time"
6)
7 
8func say(s string) {
9    for i := 0; i < 3; i++ {
10        time.Sleep(100 * time.Millisecond)
11        fmt.Println(s)
12    }
13}
14 
15func main() {
16    go say("world") // 启动新 goroutine
17    say("hello")    // 主 goroutine
18}

Channel

Channel 是 Goroutine 之间通信的管道。通过 channel，goroutine 可以安全地传递数据。

Go 的并发哲学是：不要通过共享内存来通信，而要通过通信来共享内存。

Channel 操作

1package main
2 
3import "fmt"
4 
5func main() {
6    // 创建 channel
7    ch := make(chan int)
8    
9    // 带缓冲的 channel
10    buffered := make(chan string, 2)
11    
12    // 发送和接收
13    go func() {
14        ch <- 42 // 发送
15    }()
16    
17    value := <-ch // 接收
18    fmt.Println(value)
19    
20    // 缓冲 channel
21    buffered <- "hello"
22    buffered <- "world"
23    fmt.Println(<-buffered)
24    fmt.Println(<-buffered)
25}

Channel 模式

1package main
2 
3import "fmt"
4 
5// 生产者-消费者模式
6func producer(ch chan<- int) {
7    for i := 0; i < 5; i++ {
8        ch <- i
9    }
10    close(ch)
11}
12 
13func consumer(ch <-chan int) {
14    for v := range ch {
15        fmt.Println("Received:", v)
16    }
17}
18 
19func main() {
20    ch := make(chan int)
21    
22    go producer(ch)
23    consumer(ch)
24}

Select 多路复用

select 语句让 goroutine 可以同时等待多个 channel 操作。

select 会阻塞直到某个 case 可以执行，如果多个 case 同时就绪则随机选择一个。

Select 用法

1package main
2 
3import (
4    "fmt"
5    "time"
6)
7 
8func main() {
9    ch1 := make(chan string)
10    ch2 := make(chan string)
11    
12    go func() {
13        time.Sleep(100 * time.Millisecond)
14        ch1 <- "one"
15    }()
16    
17    go func() {
18        time.Sleep(200 * time.Millisecond)
19        ch2 <- "two"
20    }()
21    
22    // 多路复用
23    for i := 0; i < 2; i++ {
24        select {
25        case msg1 := <-ch1:
26            fmt.Println("Received", msg1)
27        case msg2 := <-ch2:
28            fmt.Println("Received", msg2)
29        }
30    }
31}

超时控制

1package main
2 
3import (
4    "fmt"
5    "time"
6)
7 
8func main() {
9    ch := make(chan int)
10    
11    go func() {
12        time.Sleep(2 * time.Second)
13        ch <- 1
14    }()
15    
16    select {
17    case v := <-ch:
18        fmt.Println("Received:", v)
19    case <-time.After(1 * time.Second):
20        fmt.Println("Timeout!")
21    }
22}

Sync 包

sync 包提供了基本的同步原语，如 Mutex、WaitGroup、Once 等。

虽然 Channel 是首选的同步方式，但在某些场景下 Mutex 更高效。

WaitGroup 等待组

1package main
2 
3import (
4    "fmt"
5    "sync"
6)
7 
8func main() {
9    var wg sync.WaitGroup
10    
11    for i := 0; i < 5; i++ {
12        wg.Add(1)
13        go func(n int) {
14            defer wg.Done()
15            fmt.Println("Worker", n)
16        }(i)
17    }
18    
19    wg.Wait() // 等待所有 goroutine 完成
20    fmt.Println("All done")
21}

Mutex 互斥锁

1package main
2 
3import (
4    "fmt"
5    "sync"
6)
7 
8type SafeCounter struct {
9    mu    sync.Mutex
10    count int
11}
12 
13func (c *SafeCounter) Inc() {
14    c.mu.Lock()
15    defer c.mu.Unlock()
16    c.count++
17}
18 
19func (c *SafeCounter) Value() int {
20    c.mu.Lock()
21    defer c.mu.Unlock()
22    return c.count
23}
24 
25func main() {
26    counter := SafeCounter{}
27    var wg sync.WaitGroup
28    
29    for i := 0; i < 1000; i++ {
30        wg.Add(1)
31        go func() {
32            defer wg.Done()
33            counter.Inc()
34        }()
35    }
36    
37    wg.Wait()
38    fmt.Println(counter.Value()) // 1000
39}

Context 上下文

Context 用于在 goroutine 之间传递截止时间、取消信号和请求范围的值。

Context 是并发编程中控制 goroutine 生命周期的重要工具。

Context 使用

1package main
2 
3import (
4    "context"
5    "fmt"
6    "time"
7)
8 
9func worker(ctx context.Context) {
10    for {
11        select {
12        case <-ctx.Done():
13            fmt.Println("Worker stopped:", ctx.Err())
14            return
15        default:
16            fmt.Println("Working...")
17            time.Sleep(500 * time.Millisecond)
18        }
19    }
20}
21 
22func main() {
23    // 带超时的 context
24    ctx, cancel := context.WithTimeout(
25        context.Background(),
26        2*time.Second,
27    )
28    defer cancel()
29    
30    go worker(ctx)
31    
32    time.Sleep(3 * time.Second)
33}

1package main 2 3import ( 4 "fmt" 5 "time" 6) 7 8func say(s string) { 9 for i := 0; i < 3; i++ { 10 time.Sleep(100 * time.Millisecond) 11 fmt.Println(s) 12 } 13} 14 15func main() { 16 go say("world") // 启动新 goroutine 17 say("hello") // 主 goroutine 18}

1package main 2 3import "fmt" 4 5func main() { 6 // 创建 channel 7 ch := make(chan int) 8 9 // 带缓冲的 channel 10 buffered := make(chan string, 2) 11 12 // 发送和接收 13 go func() { 14 ch <- 42 // 发送 15 }() 16 17 value := <-ch // 接收 18 fmt.Println(value) 19 20 // 缓冲 channel 21 buffered <- "hello" 22 buffered <- "world" 23 fmt.Println(<-buffered) 24 fmt.Println(<-buffered) 25}

1package main 2 3import "fmt" 4 5// 生产者-消费者模式 6func producer(ch chan<- int) { 7 for i := 0; i < 5; i++ { 8 ch <- i 9 } 10 close(ch) 11} 12 13func consumer(ch <-chan int) { 14 for v := range ch { 15 fmt.Println("Received:", v) 16 } 17} 18 19func main() { 20 ch := make(chan int) 21 22 go producer(ch) 23 consumer(ch) 24}

1package main 2 3import ( 4 "fmt" 5 "time" 6) 7 8func main() { 9 ch1 := make(chan string) 10 ch2 := make(chan string) 11 12 go func() { 13 time.Sleep(100 * time.Millisecond) 14 ch1 <- "one" 15 }() 16 17 go func() { 18 time.Sleep(200 * time.Millisecond) 19 ch2 <- "two" 20 }() 21 22 // 多路复用 23 for i := 0; i < 2; i++ { 24 select { 25 case msg1 := <-ch1: 26 fmt.Println("Received", msg1) 27 case msg2 := <-ch2: 28 fmt.Println("Received", msg2) 29 } 30 } 31}

1package main 2 3import ( 4 "fmt" 5 "time" 6) 7 8func main() { 9 ch := make(chan int) 10 11 go func() { 12 time.Sleep(2 * time.Second) 13 ch <- 1 14 }() 15 16 select { 17 case v := <-ch: 18 fmt.Println("Received:", v) 19 case <-time.After(1 * time.Second): 20 fmt.Println("Timeout!") 21 } 22}

1package main 2 3import ( 4 "fmt" 5 "sync" 6) 7 8func main() { 9 var wg sync.WaitGroup 10 11 for i := 0; i < 5; i++ { 12 wg.Add(1) 13 go func(n int) { 14 defer wg.Done() 15 fmt.Println("Worker", n) 16 }(i) 17 } 18 19 wg.Wait() // 等待所有 goroutine 完成 20 fmt.Println("All done") 21}

1package main 2 3import ( 4 "fmt" 5 "sync" 6) 7 8type SafeCounter struct { 9 mu sync.Mutex 10 count int 11} 12 13func (c *SafeCounter) Inc() { 14 c.mu.Lock() 15 defer c.mu.Unlock() 16 c.count++ 17} 18 19func (c *SafeCounter) Value() int { 20 c.mu.Lock() 21 defer c.mu.Unlock() 22 return c.count 23} 24 25func main() { 26 counter := SafeCounter{} 27 var wg sync.WaitGroup 28 29 for i := 0; i < 1000; i++ { 30 wg.Add(1) 31 go func() { 32 defer wg.Done() 33 counter.Inc() 34 }() 35 } 36 37 wg.Wait() 38 fmt.Println(counter.Value()) // 1000 39}

1package main 2 3import ( 4 "context" 5 "fmt" 6 "time" 7) 8 9func worker(ctx context.Context) { 10 for { 11 select { 12 case <-ctx.Done(): 13 fmt.Println("Worker stopped:", ctx.Err()) 14 return 15 default: 16 fmt.Println("Working...") 17 time.Sleep(500 * time.Millisecond) 18 } 19 } 20} 21 22func main() { 23 // 带超时的 context 24 ctx, cancel := context.WithTimeout( 25 context.Background(), 26 2*time.Second, 27 ) 28 defer cancel() 29 30 go worker(ctx) 31 32 time.Sleep(3 * time.Second) 33}