Ceryn

Go 并发模式

6分钟阅读

活跃分钟 0访问量 0点赞量 0点击量 0

引入

goroutine 启动很容易，但怎么等它们全部完成？怎么保护共享数据不被同时修改？怎么在不需要时通知所有 goroutine 停下来？这些问题靠 channel 和 select 能解决一部分，但 Go 标准库提供了更趁手的工具：WaitGroup、Mutex、RWMutex 和 context 包。

正文

定义

定义

并发模式是 Go 中管理和协调 goroutine 的常用方法。WaitGroup 等待一组 goroutine 完成。Mutex 互斥锁保护共享资源，同一时刻只有一个 goroutine 能访问。RWMutex 读写锁区分读和写，多个读者可以并发访问。context 包用来传递截止时间、取消信号和请求级别的数据。

语法

// WaitGroup
var wg sync.WaitGroup
wg.Add(1)
go func() {
    defer wg.Done()
    // 做事
}()
wg.Wait()
 
// Mutex
var mu sync.Mutex
mu.Lock()
// 访问共享资源
mu.Unlock()
 
// RWMutex
var rwmu sync.RWMutex
rwmu.RLock()   // 读锁
rwmu.RUnlock()
rwmu.Lock()    // 写锁
rwmu.Unlock()
 
// context
ctx, cancel := context.WithCancel(context.Background())
defer cancel()

例子

WaitGroup 等待所有 goroutine 完成

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()
    fmt.Printf("worker %d 开始\n", id)
    time.Sleep(100 * time.Millisecond)
    fmt.Printf("worker %d 完成\n", id)
}
 
func main() {
    var wg sync.WaitGroup
 
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go worker(i, &wg)
    }
 
    wg.Wait()
    fmt.Println("全部完成")
}

wg.Add(1) 计数器加一，wg.Done() 计数器减一，wg.Wait() 阻塞直到计数器归零。

Mutex 保护共享数据

type Counter struct {
    mu    sync.Mutex
    count int
}
 
func (c *Counter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.count++
}
 
func (c *Counter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.count
}
 
func main() {
    var c Counter
    var wg sync.WaitGroup
 
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            c.Increment()
        }()
    }
 
    wg.Wait()
    fmt.Println(c.Value()) // 1000
}

不加锁的话 1000 个 goroutine 同时 count++，最终结果会小于 1000。Mutex 保证同一时刻只有一个 goroutine 修改 count。

RWMutex 读写分离

type Cache struct {
    mu   sync.RWMutex
    data map[string]string
}
 
func (c *Cache) Get(key string) (string, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    v, ok := c.data[key]
    return v, ok
}
 
func (c *Cache) Set(key, value string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.data[key] = value
}
 
func main() {
    c := &Cache{data: make(map[string]string)}
 
    c.Set("name", "Alice")
 
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            v, _ := c.Get("name")
            fmt.Println(v)
        }()
    }
    wg.Wait()
}

RWMutex 允许多个 goroutine 同时读（RLock），但写操作（Lock）是独占的。适合读多写少的场景。

context.WithCancel 取消信号

func worker(ctx context.Context, id int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("worker %d 收到取消信号\n", id)
            return
        default:
            fmt.Printf("worker %d 工作中\n", id)
            time.Sleep(100 * time.Millisecond)
        }
    }
}
 
func main() {
    ctx, cancel := context.WithCancel(context.Background())
 
    for i := 1; i <= 3; i++ {
        go worker(ctx, i)
    }
 
    time.Sleep(300 * time.Millisecond)
    cancel() // 通知所有 worker 停止
    time.Sleep(100 * time.Millisecond)
}

cancel() 被调用后，ctx.Done() 这个 channel 就会关闭，所有监听 ctx.Done() 的 goroutine 都能收到信号。

context.WithTimeout 超时控制

func fetchData(ctx context.Context) (string, error) {
    select {
    case <-time.After(2 * time.Second):
        return "数据", nil
    case <-ctx.Done():
        return "", ctx.Err()
    }
}
 
func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
    defer cancel()
 
    result, err := fetchData(ctx)
    if err != nil {
        fmt.Println("错误:", err) // context deadline exceeded
        return
    }
    fmt.Println(result)
}

context.WithTimeout() 创建一个带截止时间的 context，超过指定时间 ctx.Done() 就会触发。

context.WithValue 传递数据

type ctxKey string
 
func handler(ctx context.Context) {
    v := ctx.Value(ctxKey("requestID"))
    fmt.Println("requestID:", v)
}
 
func main() {
    ctx := context.WithValue(context.Background(), ctxKey("requestID"), "abc-123")
    handler(ctx)
}

context.WithValue() 在 context 中附带键值对数据，适合传递请求 ID、用户信息等请求级别的数据。不建议用来传大量数据。

常见写法

写法说明
wg.Add(1) / wg.Done() / wg.Wait()WaitGroup 三件套
mu.Lock() / defer mu.Unlock()Mutex 加锁解锁
rwmu.RLock() / rwmu.RUnlock()RWMutex 读锁
rwmu.Lock() / rwmu.Unlock()RWMutex 写锁
context.WithCancel(parent)创建可取消 context
context.WithTimeout(parent, d)创建带超时的 context
context.WithValue(parent, k, v)创建带数据的 context
<-ctx.Done()监听取消信号
ctx.Err()获取取消原因

特点

  • WaitGroup 用来等待一组 goroutine 完成，Add / Done / Wait 三步走
  • Mutex 保证同一时刻只有一个 goroutine 访问共享资源
  • RWMutex 区分读写，多个读者可以并发，写者独占
  • context 是 Go 并发编程的取消和超时标准方案
  • context 是树状结构，父取消则子全部取消
  • context.WithValue 只传请求级别的数据，不要滥用
  • defer mu.Unlock() 是防止忘记解锁的最佳实践

理解

WaitGroup 解决”等所有任务完成”的问题，Mutex 解决”别同时改同一个东西”的问题，context 解决”该停了就停”的问题。这四个工具组合起来，基本覆盖了 Go 并发编程中的大部分协调场景。用起来也不复杂：WaitGroup 三个方法，Mutex 加锁解锁，context 创建传递监听，记住这几套模式就够了。

引出

这些并发工具用好了能让程序又安全又高效，但如果用不好——比如忘了加锁、解锁两次——就会出现竞态条件。接下来看 Go 并发安全，学习怎么发现和避免并发安全问题。

关系图谱

反向链接