Go语言并发编程实战

前言

Go语言天生为并发而生，其简洁的并发模型让开发者可以轻松编写高性能的并发程序。本文将深入探讨Go并发编程的核心概念、常见模式和最佳实践，帮助你写出更高效、更优雅的并发代码。

一、Go并发基础

1.1 Goroutine入门

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
package main

import (
    "fmt"
    "time"
)

// 普通函数
func sayHello(name string) {
    for i := 0; i < 3; i++ {
        fmt.Printf("Hello, %s! (count: %d)\n", name, i)
        time.Sleep(100 * time.Millisecond)
    }
}

func main() {
    // 启动goroutine
    go sayHello("Alice")
    go sayHello("Bob")
    
    // 匿名函数goroutine
    go func() {
        fmt.Println("Anonymous goroutine")
    }()
    
    // 等待goroutine完成
    time.Sleep(500 * time.Millisecond)
    fmt.Println("Main function done")
}

Goroutine特点：

  • 轻量级：初始栈只有2KB
  • 调度高效：M:N调度模型
  • 通信方便：通过channel通信

1.2 Channel基础

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
// 创建channel
ch := make(chan int)           // 无缓冲channel
buffered := make(chan int, 10) // 缓冲channel

// 发送和接收
ch <- 42        // 发送
value := <-ch   // 接收
value, ok := <-ch  // 接收并检查channel是否关闭

// 关闭channel
close(ch)

// 只读/只写channel
func send(ch chan<- int) {  // 只能发送
    ch <- 1
}

func receive(ch <-chan int) {  // 只能接收
    val := <-ch
}

Channel示例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
func sum(numbers []int, resultCh chan int) {
    sum := 0
    for _, num := range numbers {
        sum += num
    }
    resultCh <- sum  // 发送结果
}

func main() {
    numbers := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
    
    resultCh := make(chan int)
    
    // 分成两部分并发计算
    go sum(numbers[:len(numbers)/2], resultCh)
    go sum(numbers[len(numbers)/2:], resultCh)
    
    // 接收两个结果
    sum1, sum2 := <-resultCh, <-resultCh
    
    fmt.Printf("Total: %d\n", sum1+sum2)
}

二、并发同步

2.1 WaitGroup

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
import (
    "fmt"
    "sync"
    "time"
)

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()  // 完成时调用Done
    
    fmt.Printf("Worker %d starting\n", id)
    time.Sleep(time.Second)
    fmt.Printf("Worker %d done\n", id)
}

func main() {
    var wg sync.WaitGroup
    
    // 启动5个worker
    for i := 1; i <= 5; i++ {
        wg.Add(1)  // 计数器+1
        go worker(i, &wg)
    }
    
    wg.Wait()  // 等待所有goroutine完成
    fmt.Println("All workers done")
}

2.2 Mutex互斥锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
type SafeCounter struct {
    mu    sync.Mutex
    count map[string]int
}

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

func (c *SafeCounter) Value(key string) int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.count[key]
}

func main() {
    counter := SafeCounter{count: make(map[string]int)}
    
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            counter.Inc("key")
        }()
    }
    
    wg.Wait()
    fmt.Println(counter.Value("key"))  // 输出: 1000
}

2.3 RWMutex读写锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
type SafeMap struct {
    mu   sync.RWMutex
    data map[string]string
}

func (m *SafeMap) Get(key string) (string, bool) {
    m.mu.RLock()         // 读锁
    defer m.mu.RUnlock()
    val, ok := m.data[key]
    return val, ok
}

func (m *SafeMap) Set(key, value string) {
    m.mu.Lock()          // 写锁
    defer m.mu.Unlock()
    m.data[key] = value
}

func main() {
    sm := SafeMap{data: make(map[string]string)}
    
    // 多个读者
    for i := 0; i < 10; i++ {
        go func(id int) {
            val, _ := sm.Get("key")
            fmt.Printf("Reader %d: %s\n", id, val)
        }(i)
    }
    
    // 一个写者
    go func() {
        sm.Set("key", "value")
    }()
    
    time.Sleep(time.Second)
}

2.4 Once确保只执行一次

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
type Config struct {
    once sync.Once
    data string
}

func (c *Config) Load() {
    c.once.Do(func() {
        fmt.Println("Loading config...")
        time.Sleep(time.Second)
        c.data = "config loaded"
    })
}

func main() {
    config := &Config{}
    
    // 多次调用，只执行一次
    for i := 0; i < 5; i++ {
        go config.Load()
    }
    
    time.Sleep(2 * time.Second)
    fmt.Println(config.data)
}

三、Channel高级用法

3.1 Select多路复用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)
    
    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "from ch1"
    }()
    
    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "from ch2"
    }()
    
    // 等待任意一个channel
    for i := 0; i < 2; i++ {
        select {
        case msg1 := <-ch1:
            fmt.Println(msg1)
        case msg2 := <-ch2:
            fmt.Println(msg2)
        case <-time.After(3 * time.Second):
            fmt.Println("timeout")
            return
        }
    }
}

3.2 超时控制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
func doWork() <-chan string {
    result := make(chan string)
    
    go func() {
        time.Sleep(2 * time.Second)
        result <- "work done"
    }()
    
    return result
}

func main() {
    result := doWork()
    
    select {
    case res := <-result:
        fmt.Println(res)
    case <-time.After(1 * time.Second):
        fmt.Println("timeout after 1 second")
    }
}

3.3 Fan-out/Fan-in模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
// Fan-out: 多个goroutine从同一个channel读取
func producer(numbers ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range numbers {
            out <- n
        }
    }()
    return out
}

// Worker: 处理数据
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// Fan-in: 合并多个channel
func merge(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    
    wg.Add(len(channels))
    
    for _, ch := range channels {
        go func(c <-chan int) {
            defer wg.Done()
            for n := range c {
                out <- n
            }
        }(ch)
    }
    
    go func() {
        wg.Wait()
        close(out)
    }()
    
    return out
}

func main() {
    // Pipeline: producer -> square -> merge
    in := producer(1, 2, 3, 4, 5)
    
    // Fan-out: 启动3个worker
    c1 := square(in)
    c2 := square(in)
    c3 := square(in)
    
    // Fan-in: 合并结果
    for n := range merge(c1, c2, c3) {
        fmt.Println(n)
    }
}

四、并发模式

4.1 生产者-消费者模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
type Task struct {
    ID   int
    Data string
}

func producer(tasks chan<- Task, count int) {
    defer close(tasks)
    
    for i := 1; i <= count; i++ {
        task := Task{
            ID:   i,
            Data: fmt.Sprintf("Task-%d", i),
        }
        tasks <- task
        fmt.Printf("Produced: %v\n", task)
        time.Sleep(100 * time.Millisecond)
    }
}

func consumer(id int, tasks <-chan Task, wg *sync.WaitGroup) {
    defer wg.Done()
    
    for task := range tasks {
        fmt.Printf("Consumer %d processing %v\n", id, task)
        time.Sleep(200 * time.Millisecond)
    }
}

func main() {
    tasks := make(chan Task, 10)
    
    // 1个生产者
    go producer(tasks, 20)
    
    // 3个消费者
    var wg sync.WaitGroup
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go consumer(i, tasks, &wg)
    }
    
    wg.Wait()
    fmt.Println("All tasks processed")
}

4.2 Worker Pool模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
type Job struct {
    ID     int
    Number int
}

type Result struct {
    Job    Job
    Sum    int
}

func worker(id int, jobs <-chan Job, results chan<- Result) {
    for job := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job.ID)
        
        // 模拟计算
        sum := 0
        for i := 1; i <= job.Number; i++ {
            sum += i
        }
        
        results <- Result{Job: job, Sum: sum}
    }
}

func main() {
    numJobs := 100
    numWorkers := 5
    
    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)
    
    // 启动worker pool
    for w := 1; w <= numWorkers; w++ {
        go worker(w, jobs, results)
    }
    
    // 发送任务
    for j := 1; j <= numJobs; j++ {
        jobs <- Job{ID: j, Number: j * 10}
    }
    close(jobs)
    
    // 收集结果
    for i := 1; i <= numJobs; i++ {
        result := <-results
        fmt.Printf("Job %d result: %d\n", result.Job.ID, result.Sum)
    }
}

4.3 Pipeline模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
// 阶段1: 生成数字
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// 阶段2: 平方
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// 阶段3: 过滤
func filter(in <-chan int, threshold int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if n > threshold {
                out <- n
            }
        }
    }()
    return out
}

func main() {
    // 构建pipeline
    numbers := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    squared := square(numbers)
    filtered := filter(squared, 50)
    
    // 消费结果
    for n := range filtered {
        fmt.Println(n)
    }
}

五、Context上下文管理

5.1 Context基础

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import "context"

func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d stopping: %v\n", id, ctx.Err())
            return
        default:
            fmt.Printf("Worker %d working\n", id)
            time.Sleep(500 * time.Millisecond)
        }
    }
}

func main() {
    // 创建可取消的context
    ctx, cancel := context.WithCancel(context.Background())
    
    // 启动多个worker
    for i := 1; i <= 3; i++ {
        go worker(ctx, i)
    }
    
    // 2秒后取消
    time.Sleep(2 * time.Second)
    cancel()
    
    // 等待goroutine退出
    time.Sleep(time.Second)
}

5.2 超时Context

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
func doRequest(ctx context.Context) error {
    result := make(chan string, 1)
    
    go func() {
        // 模拟耗时操作
        time.Sleep(2 * time.Second)
        result <- "success"
    }()
    
    select {
    case <-ctx.Done():
        return ctx.Err()
    case res := <-result:
        fmt.Println(res)
        return nil
    }
}

func main() {
    // 1秒超时
    ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
    defer cancel()
    
    if err := doRequest(ctx); err != nil {
        fmt.Printf("Request failed: %v\n", err)
    }
}

5.3 传递值

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
type key string

func worker(ctx context.Context) {
    // 获取context中的值
    if userID, ok := ctx.Value(key("userID")).(string); ok {
        fmt.Printf("Processing for user: %s\n", userID)
    }
    
    if requestID, ok := ctx.Value(key("requestID")).(string); ok {
        fmt.Printf("Request ID: %s\n", requestID)
    }
}

func main() {
    ctx := context.Background()
    ctx = context.WithValue(ctx, key("userID"), "user123")
    ctx = context.WithValue(ctx, key("requestID"), "req456")
    
    go worker(ctx)
    
    time.Sleep(time.Second)
}

六、并发安全

6.1 数据竞争检测

1
2
3
4
# 使用race detector
go run -race main.go
go build -race
go test -race
1
2
3
4
5
6
7
8
9
10
11
12
13
14
// 数据竞争示例
var counter int

func increment() {
    counter++  // 数据竞争！
}

func main() {
    for i := 0; i < 1000; i++ {
        go increment()
    }
    time.Sleep(time.Second)
    fmt.Println(counter)  // 结果不确定
}

6.2 原子操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import "sync/atomic"

var counter int64

func increment() {
    atomic.AddInt64(&counter, 1)
}

func main() {
    for i := 0; i < 1000; i++ {
        go increment()
    }
    
    time.Sleep(time.Second)
    fmt.Println(atomic.LoadInt64(&counter))  // 输出: 1000
}

七、性能优化

7.1 避免Goroutine泄露

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// ❌ 错误：goroutine永远不会退出
func leak() {
    ch := make(chan int)
    go func() {
        val := <-ch  // 永远阻塞
        fmt.Println(val)
    }()
}

// ✅ 正确：使用context或关闭channel
func noLeak() {
    ch := make(chan int)
    done := make(chan bool)
    
    go func() {
        select {
        case val := <-ch:
            fmt.Println(val)
        case <-done:
            return
        }
    }()
    
    close(done)  // 确保goroutine退出
}

7.2 控制并发数量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
func limitConcurrency(maxGoroutines int) {
    sem := make(chan struct{}, maxGoroutines)
    
    tasks := make([]int, 100)
    for i := range tasks {
        tasks[i] = i
    }
    
    var wg sync.WaitGroup
    for _, task := range tasks {
        wg.Add(1)
        
        // 获取信号量
        sem <- struct{}{}
        
        go func(t int) {
            defer wg.Done()
            defer func() { <-sem }()  // 释放信号量
            
            fmt.Printf("Processing task %d\n", t)
            time.Sleep(100 * time.Millisecond)
        }(task)
    }
    
    wg.Wait()
}

八、实战案例

8.1 并发网络爬虫

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
type Fetcher interface {
    Fetch(url string) (urls []string, err error)
}

func crawl(url string, depth int, fetcher Fetcher, 
           visited map[string]bool, mu *sync.Mutex, wg *sync.WaitGroup) {
    defer wg.Done()
    
    if depth <= 0 {
        return
    }
    
    mu.Lock()
    if visited[url] {
        mu.Unlock()
        return
    }
    visited[url] = true
    mu.Unlock()
    
    urls, err := fetcher.Fetch(url)
    if err != nil {
        fmt.Println(err)
        return
    }
    
    fmt.Printf("found: %s\n", url)
    
    for _, u := range urls {
        wg.Add(1)
        go crawl(u, depth-1, fetcher, visited, mu, wg)
    }
}

func Crawl(url string, depth int, fetcher Fetcher) {
    visited := make(map[string]bool)
    var mu sync.Mutex
    var wg sync.WaitGroup
    
    wg.Add(1)
    go crawl(url, depth, fetcher, visited, &mu, &wg)
    
    wg.Wait()
}

九、最佳实践

核心原则

  1. 不要通过共享内存来通信，而要通过通信来共享内存
  2. 总是使用defer关闭channel和释放锁
  3. 避免goroutine泄露，确保所有goroutine都能退出
  4. 使用context管理goroutine生命周期
  5. 控制并发数量，避免创建过多goroutine

常见陷阱

1
2
3
4
5
6
7
8
9
10
11
12
13
// ❌ 错误：循环变量捕获
for i := 0; i < 10; i++ {
    go func() {
        fmt.Println(i)  // 可能全部输出10
    }()
}

// ✅ 正确
for i := 0; i < 10; i++ {
    go func(i int) {
        fmt.Println(i)
    }(i)
}

十、总结

Go的并发模型让并发编程变得简单而优雅：

核心概念

  • Goroutine：轻量级协程
  • Channel：通信和同步
  • Select：多路复用
  • Sync包：低级同步原语

并发模式

  • Pipeline
  • Fan-out/Fan-in
  • Worker Pool
  • 生产者-消费者

掌握Go并发编程，让你的程序飞起来！

参考资料