---
id: 01933e13f963942e4c9bc8c3c89fbc2b
author: snowmerak
title: Go 并发入门包
description: 本文介绍用于管理 Go 并发的各种工具和技术,包括 goroutine、channel、mutex 等。
language: zh
date: 2024-10-18T05:51:30.010680581Z
path: /blog/posts/go-concurrency-starter-pack-z221a3399
category: concurrency
---

## 概述

### 简短介绍

Go 语言提供了许多用于管理并发的工具。在本文中，我将介绍其中一部分工具及一些技巧。

### 协程？

goroutine 是 Go 语言所支持的一种新型并发模型。通常，程序为了同时执行多项任务，会从操作系统获取 OS 线程，并根据核心数量并行地执行任务。而要实现更小粒度的并发，则需要在用户空间创建绿色线程，使多个绿色线程在一个 OS 线程内轮流执行任务。不过，goroutine 将这种形式的绿色线程做得更加小巧高效。这种 goroutine 占用的内存比线程更少，并且能够比线程更快地被创建和切换。

使用 goroutine 只需要简单地使用 `go` 关键字。这使得在编写程序的过程中，能够直观地将同步代码以异步方式执行。

```go
package main

import (
    "fmt"
    "time"
)

func main() {
    ch := make(chan struct{})
    go func() {
        defer close(ch)
        time.Sleep(1 * time.Second)
        fmt.Println("Hello, World!")
    }()

    fmt.Println("Waiting for goroutine...")
    for range ch {}
}
```

这段代码将一段先休眠 1 秒再输出 `Hello, World!` 的同步代码改写为异步流程。当前的示例较为简单，但如果将稍微复杂的代码从同步改写为异步，相比 async await 或 promise 等既有方式，代码的可读性、可见性和可理解性会更好。

然而，在许多情况下，如果不理解这种将同步代码简单地异步调用的流程，以及类似 `fork & join` 的流程（类似于分治的流程），就可能写出不良的 goroutine 代码。针对这种情况，本文将介绍一些方法和技巧。

## 并发管理

### context

作为第一个管理技术出现 `context` 或许令人意外。但在 Go 语言中，`context` 不仅仅是简单的取消功能，它在管理整个工作树方面发挥着卓越的作用。如果有不熟悉的读者，我将简要介绍该包。

```go
package main

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    go func() {
        <-ctx.Done()
        fmt.Println("Context is done!")
    }()

    time.Sleep(1 * time.Second)

    cancel()

    time.Sleep(1 * time.Second)
}
```

上述代码使用 `context`，在 1 秒后输出 `Context is done!`。`context` 可以通过 `Done()` 方法确认是否已取消，并提供 `WithCancel`、`WithTimeout`、`WithDeadline`、`WithValue` 等方法，以实现多种取消方式。

我们来编写一个简单的示例。假设您使用 `aggregator` 模式编写代码，以获取 `user`、`post` 和 `comment`，并且要求所有请求必须在 2 秒内完成，那么可以这样编写。

```go
package main

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second)
    defer cancel()

    ch := make(chan struct{})
    go func() {
        defer close(ch)
        user := getUser(ctx)
        post := getPost(ctx)
        comment := getComment(ctx)

        fmt.Println(user, post, comment)
    }()

    select {
    case <-ctx.Done():
        fmt.Println("Timeout!")
    case <-ch:
        fmt.Println("All data is fetched!")
    }
}
```

上述代码如果在 2 秒内未能获取所有数据，则输出 `Timeout!`；如果获取了所有数据，则输出 `All data is fetched!`。以这种方式使用 `context`，即使在多个 goroutine 运行的代码中，也能轻松管理取消和超时。

与 context 相关的各种函数和方法可以在 [godoc context](https://pkg.go.dev/context) 中查看。希望您能够学习简单的部分，并方便地加以使用。

### channel

#### unbuffered channel

`channel` 是用于 goroutine 之间通信的工具。`channel` 可以通过 `make(chan T)` 创建，其中 `T` 是该 `channel` 所传递数据的类型。`channel` 可以使用 `<-` 收发数据，并可以使用 `close` 关闭。

```go
package main

func main() {
    ch := make(chan int)
    go func() {
        ch <- 1
        ch <- 2
        close(ch)
    }()

    for i := range ch {
        fmt.Println(i)
    }
}
```

上述代码使用 `channel` 输出 1 和 2。该代码仅展示了向 `channel` 发送和接收值的基本操作，但 `channel` 还提供更多功能。首先，我们来了解 `buffered channel` 与 `unbuffered channel`。在此之前需要说明，上面的示例使用的是 `unbuffered channel`，向通道发送数据和接收数据的操作必须同时进行。如果这些操作不同时发生，则可能导致死锁。

#### buffered channel

如果上述代码不是简单输出，而是两个执行繁重任务的进程呢？如果第二个进程在读取并处理数据时长时间阻塞，那么第一个进程也将在此期间停止。为了防止这种情况，我们可以使用 `buffered channel`。

```go
package main

func main() {
    ch := make(chan int, 2)
    go func() {
        ch <- 1
        ch <- 2
        close(ch)
    }()

    for i := range ch {
        fmt.Println(i)
    }
}
```

上述代码使用 `buffered channel` 输出 1 和 2。在该代码中，通过使用 `buffered channel`，向 `channel` 发送数据和接收数据的操作无需同时进行。这样为通道设置缓冲区后，就会留出相应长度的余量，从而可以防止由于后续任务的影响而产生的任务延迟。

#### select

在处理多个通道时，使用 `select` 语法可以轻松实现 `fan-in` 结构。

```go
package main

import (
    "fmt"
    "time"
)

func main() {
    ch1 := make(chan int, 10)
    ch2 := make(chan int, 10)
    ch3 := make(chan int, 10)

    go func() {
        for {
            ch1 <- 1
            time.Sleep(1 * time.Second)
        }
    }()
    go func() {
        for {
            ch2 <- 2
            time.Sleep(2 * time.Second)
        }
    }()
    go func() {
        for {
            ch3 <- 3
            time.Sleep(3 * time.Second)
        }
    }()

    for i := 0; i < 3; i++ {
        select {
        case v := <-ch1:
            fmt.Println(v)
        case v := <-ch2:
            fmt.Println(v)
        case v := <-ch3:
            fmt.Println(v)
        }
    }
}
```

上述代码创建了三个定期传递 1、2、3 的通道，并使用 `select` 从通道中接收值并输出。以这种方式使用 `select`，可以同时从多个通道接收数据，并且按照值到达的顺序进行处理。

#### for range

`channel` 可以使用 `for range` 轻松接收数据。对 `channel` 使用 `for range` 时，每当该通道中有数据添加就会执行一次，当通道关闭时则结束循环。

```go
package main

func main() {
    ch := make(chan int)
    go func() {
        ch <- 1
        ch <- 2
        close(ch)
    }()

    for i := range ch {
        fmt.Println(i)
    }
}
```

上述代码使用 `channel` 输出 1 和 2。在该代码中，使用 `for range` 在每次有数据添加到通道时接收数据并输出。当通道关闭时，循环结束。

如前面多次示例所示，此语法也可用作简单的同步手段。

```go
package main

func main() {
    ch := make(chan struct{})
    go func() {
        defer close(ch)
        time.Sleep(1 * time.Second)
        fmt.Println("Hello, World!")
    }()

    fmt.Println("Waiting for goroutine...")
    for range ch {}
}
```

上述代码在休眠 1 秒后输出 `Hello, World!`。在该代码中，使用 `channel` 将同步代码改写为异步代码。以这种方式使用 `channel`，可以轻松地将同步代码改写为异步代码，并设置 `join` 节点。

#### etc

1. 向 nil channel 发送或从中接收数据，可能陷入无限阻塞并导致死锁。
2. 关闭通道后再发送数据，将引发 panic。
3. 即使不显式关闭通道，GC 在回收时也会关闭通道。

### mutex

#### spinlock

`spinlock` 是一种通过循环不断尝试获取锁的同步方法。在 Go 语言中，可以使用指针较容易地实现自旋锁。

```go
package spinlock

import (
    "runtime"
    "sync/atomic"
)

type SpinLock struct {
    lock uintptr
}

func (s *SpinLock) Lock() {
    for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) {
        runtime.Gosched()
    }
}

func (s *SpinLock) Unlock() {
    atomic.StoreUintptr(&s.lock, 0)
}

func NewSpinLock() *SpinLock {
    return &SpinLock{}
}
```

上述代码是 `spinlock` 包的实现。该代码使用 `sync/atomic` 包实现了 `SpinLock`。`Lock` 方法使用 `atomic.CompareAndSwapUintptr` 尝试获取锁，`Unlock` 方法使用 `atomic.StoreUintptr` 释放锁。由于此方式会不停地尝试获取锁，在获得锁之前会持续占用 CPU，因此可能陷入无限循环。所以，`spinlock` 最好用于简单的同步，或仅用于持续时间很短的场景。

#### sync.Mutex

`mutex` 是用于 goroutine 之间同步的工具。`sync` 包提供的 `mutex` 提供 `Lock`、`Unlock`、`RLock`、`RUnlock` 等方法。`mutex` 可以通过 `sync.Mutex` 创建，也可以使用 `sync.RWMutex` 来使用读写锁。

```go
package main

import (
    "sync"
)

func main() {
    var mu sync.Mutex
    var count int

    go func() {
        mu.Lock()
        count++
        mu.Unlock()
    }()

    mu.Lock()
    count++
    mu.Unlock()

    println(count)
}
```

在上述代码中，两个 goroutine 几乎同时访问同一个 `count` 变量。此时，如果使用 `mutex` 将访问 `count` 变量的代码设为临界区，就可以防止对 `count` 变量的并发访问。这样，无论运行多少次，该代码都会始终输出 `2`。

#### sync.RWMutex

`sync.RWMutex` 是一种可以区分读锁和写锁使用的 `mutex`。可以使用 `RLock`、`RUnlock` 方法来加锁和解锁读锁。

```go
package cmap

import (
    "sync"
)

type ConcurrentMap[K comparable, V any] struct {
    sync.RWMutex
    data map[K]V
}

func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) {
    m.RLock()
    defer m.RUnlock()

    value, ok := m.data[key]
    return value, ok
}

func (m *ConcurrentMap[K, V]) Set(key K, value V) {
    m.Lock()
    defer m.Unlock()

    m.data[key] = value
}
```

上述代码使用 `sync.RWMutex` 实现了 `ConcurrentMap`。在该代码中，`Get` 方法加读锁，`Set` 方法加写锁，从而可以安全地访问和修改 `data` 映射。之所以需要读锁，是因为在大量仅涉及读取操作的情况下，可以不加写锁，只加读锁，使多个 goroutine 能够同时执行读取操作。通过这种方式，在不需要改变状态、无需加写锁的情况下，仅加读锁即可提升性能。

#### fakelock

`fakelock` 是一种实现 `sync.Locker` 的简单技巧。该结构体提供与 `sync.Mutex` 相同的方法，但实际上不执行任何操作。

```go
package fakelock

type FakeLock struct{}

func (f *FakeLock) Lock() {}

func (f *FakeLock) Unlock() {}
```

上述代码是 `fakelock` 包的实现。该包实现了 `sync.Locker`，提供 `Lock`、`Unlock` 方法，但实际上不执行任何操作。为什么需要这样的代码，我将在合适的时机再作说明。

### waitgroup

#### sync.WaitGroup

`sync.WaitGroup` 是一种用于等待所有 goroutine 任务完成的工具。它提供 `Add`、`Done`、`Wait` 方法，其中 `Add` 方法用于增加 goroutine 的数量，`Done` 方法用于通知某个 goroutine 的任务已完成。然后，`Wait` 方法会等待所有 goroutine 的任务完成。

```go
package main

import (
    "sync"
    "sync/atomic"
)

func main() {
    wg := sync.WaitGroup{}
    c := atomic.Int64{}

    for i := 0; i < 100 ; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            c.Add(1)
        }()
    }

    wg.Wait()
    println(c.Load())
}
```

上述代码使用 `sync.WaitGroup`，让 100 个 goroutine 同时对变量 `c` 进行加法运算。在该代码中，使用 `sync.WaitGroup` 等待所有 goroutine 完成后，输出 `c` 变量累加的结果。在对少量任务进行 `fork & join` 时，仅使用通道就已足够；但在对大量任务进行 `fork & join` 时，使用 `sync.WaitGroup` 也是不错的选择。

#### with slice

与切片一同使用时，`waitgroup` 可以成为无需加锁即可管理出色并发执行任务的工具。

```go
package main

import (
	"fmt"
	"sync"
    "rand"
)

func main() {
	var wg sync.WaitGroup
	arr := [10]int{}

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()

			arr[id] = rand.Intn(100)
		}(i)
	}

	wg.Wait()
	fmt.Println("Done")

    for i, v := range arr {
        fmt.Printf("arr[%d] = %d\n", i, v)
    }
}
```

上述代码仅使用 `waitgroup`，让每个 goroutine 同时生成 10 个随机整数，并将其保存到分配的索引中。在该代码中，使用 `waitgroup` 等待所有 goroutine 完成后，输出 `Done`。通过这种方式使用 `waitgroup`，多个 goroutine 可以同时执行任务，在所有 goroutine 完成之前无需加锁即可保存数据，并在任务结束后统一进行后处理。

#### golang.org/x/sync/errgroup.ErrGroup

`errgroup` 是对 `sync.WaitGroup` 的扩展包。与 `sync.WaitGroup` 不同，`errgroup` 在任一 goroutine 的任务发生错误时，会取消所有 goroutine 并返回该错误。

```go
package main

import (
    "context"
    "fmt"
    "golang.org/x/sync/errgroup"
)

func main() {
    g, ctx := errgroup.WithContext(context.Background())
    _ = ctx

    for i := 0; i < 10; i++ {
        i := i
        g.Go(func() error {
            if i == 5 {
                return fmt.Errorf("error")
            }
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        fmt.Println(err)
    }
}
```

上述代码使用 `errgroup` 创建 10 个 goroutine，并在第 5 个 goroutine 中引发错误。为了展示错误发生的情况，我们有意在第五个 goroutine 中引发错误。但在实际使用时，应使用 `errgroup` 创建 goroutine，并针对各 goroutine 中发生错误的情况进行各种后处理。

### once

这是一种只应执行一次的代码的执行工具。可以通过下面的构造函数来执行相关代码。

```go
func OnceFunc(f func()) func()
func OnceValue[T any](f func() T) func() T
func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2)
```

#### OnceFunc

`OnceFunc` 仅保证该函数在整个程序中只执行一次。

```go
package main

import "sync"

func main() {
    once := sync.OnceFunc(func() {
        println("Hello, World!")
    })

    once()
    once()
    once()
    once()
    once()
}
```

上述代码使用 `sync.OnceFunc` 输出 `Hello, World!`。在该代码中，使用 `sync.OnceFunc` 创建 `once` 函数，即使多次调用 `once` 函数，`Hello, World!` 也只会输出一次。

#### OnceValue

`OnceValue` 并非仅保证该函数只执行一次，而是会保存该函数的返回值，在再次调用时返回已保存的值。

```go
package main

import "sync"

func main() {
    c := 0
    once := sync.OnceValue(func() int {
        c += 1
        return c
    })

    println(once())
    println(once())
    println(once())
    println(once())
    println(once())
}
```

上述代码使用 `sync.OnceValue` 使变量 `c` 每次增加 1。在该代码中，使用 `sync.OnceValue` 创建 `once` 函数，即使多次调用 `once` 函数，变量 `c` 也只会增加一次，并返回增加后的值 1。

#### OnceValues

`OnceValues` 的作用与 `OnceValue` 相同，但可以返回多个值。

```go
package main

import "sync"

func main() {
    c := 0
    once := sync.OnceValues(func() (int, int) {
        c += 1
        return c, c
    })

    a, b := once()
    println(a, b)
    a, b = once()
    println(a, b)
    a, b = once()
    println(a, b)
    a, b = once()
    println(a, b)
    a, b = once()
    println(a, b)
}
```

上述代码使用 `sync.OnceValues` 使变量 `c` 每次增加 1。在该代码中，使用 `sync.OnceValues` 创建 `once` 函数，即使多次调用 `once` 函数，变量 `c` 也只会增加一次，并返回增加后的值 1。

### atomic

`atomic` 包是提供原子操作的包。`atomic` 包提供 `Add`、`CompareAndSwap`、`Load`、`Store`、`Swap` 等方法，但近来更推荐使用 `Int64`、`Uint64`、`Pointer` 等类型。

```go
package main

import (
    "sync"
    "sync/atomic"
)

func main() {
    wg := sync.WaitGroup{}
    c := atomic.Int64{}

    for i := 0; i < 100 ; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            c.Add(1)
        }()
    }

    wg.Wait()
    println(c.Load())
}
```

这是之前使用过的示例。该代码使用 `atomic.Int64` 类型以原子方式增加变量 `c`。通过 `Add` 方法和 `Load` 方法，可以原子地增加变量并读取变量的值。此外，可以使用 `Store` 方法存储值，使用 `Swap` 方法交换值，并使用 `CompareAndSwap` 方法比较值，若符合条件则进行替换。

### cond

#### sync.Cond

`cond` 包是提供条件变量的包。`cond` 包可以通过 `sync.Cond` 创建，并提供 `Wait`、`Signal`、`Broadcast` 方法。

```go
package main

import (
    "sync"
)

func main() {
    c := sync.NewCond(&sync.Mutex{})
    ready := false

    go func() {
        c.L.Lock()
        ready = true
        c.Signal()
        c.L.Unlock()
    }()

    c.L.Lock()
    for !ready {
        c.Wait()
    }
    c.L.Unlock()

    println("Ready!")
}
```

上述代码使用 `sync.Cond` 等待变量 `ready` 变为 `true`。在该代码中，使用 `sync.Cond` 等待 `ready` 变量变为 `true` 后，输出 `Ready!`。以这种方式使用 `sync.Cond`，可以让多个 goroutine 同时等待某个特定条件得到满足。

利用此特性，可以实现一个简单的 `queue`。

```go
package queue

import (
    "sync"
    "sync/atomic"
)

type Node[T any] struct {
    Value T
    Next  *Node[T]
}

type Queue[T any] struct {
    sync.Mutex
    Cond *sync.Cond
    Head *Node[T]
    Tail *Node[T]
    Len  int
}

func New[T any]() *Queue[T] {
    q := &Queue[T]{}
    q.Cond = sync.NewCond(&q.Mutex)
    return q
}

func (q *Queue[T]) Push(value T) {
    q.Lock()
    defer q.Unlock()

    node := &Node[T]{Value: value}
    if q.Len == 0 {
        q.Head = node
        q.Tail = node
    } else {
        q.Tail.Next = node
        q.Tail = node
    }
    q.Len++
    q.Cond.Signal()
}

func (q *Queue[T]) Pop() T {
    q.Lock()
    defer q.Unlock()

    for q.Len == 0 {
        q.Cond.Wait()
    }

    node := q.Head
    q.Head = q.Head.Next
    q.Len--
    return node.Value
}
```

如此使用 `sync.Cond`，就不必通过 `spin-lock` 占用大量 CPU，而是可以高效地等待，并在条件满足时恢复运行。

### semaphore

#### golang.org/x/sync/semaphore.Semaphore

`semaphore` 包是提供信号量的包。`semaphore` 包可以通过 `golang.org/x/sync/semaphore.Semaphore` 创建，并提供 `Acquire`、`Release`、`TryAcquire` 方法。

```go
package main

import (
    "context"
    "fmt"
    "golang.org/x/sync/semaphore"
)

func main() {
    s := semaphore.NewWeighted(1)

    if s.TryAcquire(1) {
        fmt.Println("Acquired!")
    } else {
        fmt.Println("Not Acquired!")
    }

    s.Release(1)
}
```

上述代码使用 `semaphore` 创建信号量，并通过 `Acquire` 方法获取信号量，再通过 `Release` 方法释放信号量。该代码展示了使用 `semaphore` 获取和释放信号量的方法。

## 结语

基本内容到这里应该就足够了。希望您能够基于本文的内容理解如何使用 goroutine 管理并发，并能够在实际中加以运用。希望本文对您有所帮助。谢谢。