GoSudaGo 并发入门包
概述
简短介绍
Go 语言提供了许多用于管理并发的工具。在本文中，我将介绍其中一部分工具及一些技巧。
协程？
goroutine 是 Go 语言所支持的一种新型并发模型。通常，程序为了同时执行多项任务，会从操作系统获取 OS 线程，并根据核心数量并行地执行任务。而要实现更小粒度的并发，则需要在用户空间创建绿色线程，使多个绿色线程在一个 OS 线程内轮流执行任务。不过，goroutine 将这种形式的绿色线程做得更加小巧高效。这种 goroutine 占用的内存比线程更少，并且能够比线程更快地被创建和切换。
使用 goroutine 只需要简单地使用 go 关键字。这使得在编写程序的过程中，能够直观地将同步代码以异步方式执行。
1package main
2
3import (
4 "fmt"
5 "time"
6)
7
8func main() {
9 ch := make(chan struct{})
10 go func() {
11 defer close(ch)
12 time.Sleep(1 * time.Second)
13 fmt.Println("Hello, World!")
14 }()
15
16 fmt.Println("Waiting for goroutine...")
17 for range ch {}
18}
这段代码将一段先休眠 1 秒再输出 Hello, World! 的同步代码改写为异步流程。当前的示例较为简单，但如果将稍微复杂的代码从同步改写为异步，相比 async await 或 promise 等既有方式，代码的可读性、可见性和可理解性会更好。
然而，在许多情况下，如果不理解这种将同步代码简单地异步调用的流程，以及类似 fork & join 的流程（类似于分治的流程），就可能写出不良的 goroutine 代码。针对这种情况，本文将介绍一些方法和技巧。
并发管理
context
作为第一个管理技术出现 context 或许令人意外。但在 Go 语言中，context 不仅仅是简单的取消功能，它在管理整个工作树方面发挥着卓越的作用。如果有不熟悉的读者，我将简要介绍该包。
1package main
2
3func main() {
4 ctx, cancel := context.WithCancel(context.Background())
5 defer cancel()
6
7 go func() {
8 <-ctx.Done()
9 fmt.Println("Context is done!")
10 }()
11
12 time.Sleep(1 * time.Second)
13
14 cancel()
15
16 time.Sleep(1 * time.Second)
17}
上述代码使用 context，在 1 秒后输出 Context is done!。context 可以通过 Done() 方法确认是否已取消，并提供 WithCancel、WithTimeout、WithDeadline、WithValue 等方法，以实现多种取消方式。
我们来编写一个简单的示例。假设您使用 aggregator 模式编写代码，以获取 user、post 和 comment，并且要求所有请求必须在 2 秒内完成，那么可以这样编写。
1package main
2
3func main() {
4 ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second)
5 defer cancel()
6
7 ch := make(chan struct{})
8 go func() {
9 defer close(ch)
10 user := getUser(ctx)
11 post := getPost(ctx)
12 comment := getComment(ctx)
13
14 fmt.Println(user, post, comment)
15 }()
16
17 select {
18 case <-ctx.Done():
19 fmt.Println("Timeout!")
20 case <-ch:
21 fmt.Println("All data is fetched!")
22 }
23}
上述代码如果在 2 秒内未能获取所有数据，则输出 Timeout!；如果获取了所有数据，则输出 All data is fetched!。以这种方式使用 context，即使在多个 goroutine 运行的代码中，也能轻松管理取消和超时。
与 context 相关的各种函数和方法可以在 godoc context 中查看。希望您能够学习简单的部分，并方便地加以使用。
channel
unbuffered channel
channel 是用于 goroutine 之间通信的工具。channel 可以通过 make(chan T) 创建，其中 T 是该 channel 所传递数据的类型。channel 可以使用 <- 收发数据，并可以使用 close 关闭。
1package main
2
3func main() {
4 ch := make(chan int)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
上述代码使用 channel 输出 1 和 2。该代码仅展示了向 channel 发送和接收值的基本操作，但 channel 还提供更多功能。首先，我们来了解 buffered channel 与 unbuffered channel。在此之前需要说明，上面的示例使用的是 unbuffered channel，向通道发送数据和接收数据的操作必须同时进行。如果这些操作不同时发生，则可能导致死锁。
buffered channel
如果上述代码不是简单输出，而是两个执行繁重任务的进程呢？如果第二个进程在读取并处理数据时长时间阻塞，那么第一个进程也将在此期间停止。为了防止这种情况，我们可以使用 buffered channel。
1package main
2
3func main() {
4 ch := make(chan int, 2)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
上述代码使用 buffered channel 输出 1 和 2。在该代码中，通过使用 buffered channel，向 channel 发送数据和接收数据的操作无需同时进行。这样为通道设置缓冲区后，就会留出相应长度的余量，从而可以防止由于后续任务的影响而产生的任务延迟。
select
在处理多个通道时，使用 select 语法可以轻松实现 fan-in 结构。
1package main
2
3import (
4 "fmt"
5 "time"
6)
7
8func main() {
9 ch1 := make(chan int, 10)
10 ch2 := make(chan int, 10)
11 ch3 := make(chan int, 10)
12
13 go func() {
14 for {
15 ch1 <- 1
16 time.Sleep(1 * time.Second)
17 }
18 }()
19 go func() {
20 for {
21 ch2 <- 2
22 time.Sleep(2 * time.Second)
23 }
24 }()
25 go func() {
26 for {
27 ch3 <- 3
28 time.Sleep(3 * time.Second)
29 }
30 }()
31
32 for i := 0; i < 3; i++ {
33 select {
34 case v := <-ch1:
35 fmt.Println(v)
36 case v := <-ch2:
37 fmt.Println(v)
38 case v := <-ch3:
39 fmt.Println(v)
40 }
41 }
42}
上述代码创建了三个定期传递 1、2、3 的通道，并使用 select 从通道中接收值并输出。以这种方式使用 select，可以同时从多个通道接收数据，并且按照值到达的顺序进行处理。
for range
channel 可以使用 for range 轻松接收数据。对 channel 使用 for range 时，每当该通道中有数据添加就会执行一次，当通道关闭时则结束循环。
1package main
2
3func main() {
4 ch := make(chan int)
5 go func() {
6 ch <- 1
7 ch <- 2
8 close(ch)
9 }()
10
11 for i := range ch {
12 fmt.Println(i)
13 }
14}
上述代码使用 channel 输出 1 和 2。在该代码中，使用 for range 在每次有数据添加到通道时接收数据并输出。当通道关闭时，循环结束。
如前面多次示例所示，此语法也可用作简单的同步手段。
1package main
2
3func main() {
4 ch := make(chan struct{})
5 go func() {
6 defer close(ch)
7 time.Sleep(1 * time.Second)
8 fmt.Println("Hello, World!")
9 }()
10
11 fmt.Println("Waiting for goroutine...")
12 for range ch {}
13}
上述代码在休眠 1 秒后输出 Hello, World!。在该代码中，使用 channel 将同步代码改写为异步代码。以这种方式使用 channel，可以轻松地将同步代码改写为异步代码，并设置 join 节点。
etc
- 向 nil channel 发送或从中接收数据，可能陷入无限阻塞并导致死锁。
- 关闭通道后再发送数据，将引发 panic。
- 即使不显式关闭通道，GC 在回收时也会关闭通道。
mutex
spinlock
spinlock 是一种通过循环不断尝试获取锁的同步方法。在 Go 语言中，可以使用指针较容易地实现自旋锁。
1package spinlock
2
3import (
4 "runtime"
5 "sync/atomic"
6)
7
8type SpinLock struct {
9 lock uintptr
10}
11
12func (s *SpinLock) Lock() {
13 for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) {
14 runtime.Gosched()
15 }
16}
17
18func (s *SpinLock) Unlock() {
19 atomic.StoreUintptr(&s.lock, 0)
20}
21
22func NewSpinLock() *SpinLock {
23 return &SpinLock{}
24}
上述代码是 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 来使用读写锁。
1package main
2
3import (
4 "sync"
5)
6
7func main() {
8 var mu sync.Mutex
9 var count int
10
11 go func() {
12 mu.Lock()
13 count++
14 mu.Unlock()
15 }()
16
17 mu.Lock()
18 count++
19 mu.Unlock()
20
21 println(count)
22}
在上述代码中，两个 goroutine 几乎同时访问同一个 count 变量。此时，如果使用 mutex 将访问 count 变量的代码设为临界区，就可以防止对 count 变量的并发访问。这样，无论运行多少次，该代码都会始终输出 2。
sync.RWMutex
sync.RWMutex 是一种可以区分读锁和写锁使用的 mutex。可以使用 RLock、RUnlock 方法来加锁和解锁读锁。
1package cmap
2
3import (
4 "sync"
5)
6
7type ConcurrentMap[K comparable, V any] struct {
8 sync.RWMutex
9 data map[K]V
10}
11
12func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) {
13 m.RLock()
14 defer m.RUnlock()
15
16 value, ok := m.data[key]
17 return value, ok
18}
19
20func (m *ConcurrentMap[K, V]) Set(key K, value V) {
21 m.Lock()
22 defer m.Unlock()
23
24 m.data[key] = value
25}
上述代码使用 sync.RWMutex 实现了 ConcurrentMap。在该代码中，Get 方法加读锁，Set 方法加写锁，从而可以安全地访问和修改 data 映射。之所以需要读锁，是因为在大量仅涉及读取操作的情况下，可以不加写锁，只加读锁，使多个 goroutine 能够同时执行读取操作。通过这种方式，在不需要改变状态、无需加写锁的情况下，仅加读锁即可提升性能。
fakelock
fakelock 是一种实现 sync.Locker 的简单技巧。该结构体提供与 sync.Mutex 相同的方法，但实际上不执行任何操作。
1package fakelock
2
3type FakeLock struct{}
4
5func (f *FakeLock) Lock() {}
6
7func (f *FakeLock) Unlock() {}
上述代码是 fakelock 包的实现。该包实现了 sync.Locker，提供 Lock、Unlock 方法，但实际上不执行任何操作。为什么需要这样的代码，我将在合适的时机再作说明。
waitgroup
sync.WaitGroup
sync.WaitGroup 是一种用于等待所有 goroutine 任务完成的工具。它提供 Add、Done、Wait 方法，其中 Add 方法用于增加 goroutine 的数量，Done 方法用于通知某个 goroutine 的任务已完成。然后，Wait 方法会等待所有 goroutine 的任务完成。
1package main
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8func main() {
9 wg := sync.WaitGroup{}
10 c := atomic.Int64{}
11
12 for i := 0; i < 100 ; i++ {
13 wg.Add(1)
14 go func() {
15 defer wg.Done()
16 c.Add(1)
17 }()
18 }
19
20 wg.Wait()
21 println(c.Load())
22}
上述代码使用 sync.WaitGroup，让 100 个 goroutine 同时对变量 c 进行加法运算。在该代码中，使用 sync.WaitGroup 等待所有 goroutine 完成后，输出 c 变量累加的结果。在对少量任务进行 fork & join 时，仅使用通道就已足够；但在对大量任务进行 fork & join 时，使用 sync.WaitGroup 也是不错的选择。
with slice
与切片一同使用时，waitgroup 可以成为无需加锁即可管理出色并发执行任务的工具。
1package main
2
3import (
4 "fmt"
5 "sync"
6 "rand"
7)
8
9func main() {
10 var wg sync.WaitGroup
11 arr := [10]int{}
12
13 for i := 0; i < 10; i++ {
14 wg.Add(1)
15 go func(id int) {
16 defer wg.Done()
17
18 arr[id] = rand.Intn(100)
19 }(i)
20 }
21
22 wg.Wait()
23 fmt.Println("Done")
24
25 for i, v := range arr {
26 fmt.Printf("arr[%d] = %d\n", i, v)
27 }
28}
上述代码仅使用 waitgroup，让每个 goroutine 同时生成 10 个随机整数，并将其保存到分配的索引中。在该代码中，使用 waitgroup 等待所有 goroutine 完成后，输出 Done。通过这种方式使用 waitgroup，多个 goroutine 可以同时执行任务，在所有 goroutine 完成之前无需加锁即可保存数据，并在任务结束后统一进行后处理。
golang.org/x/sync/errgroup.ErrGroup
errgroup 是对 sync.WaitGroup 的扩展包。与 sync.WaitGroup 不同，errgroup 在任一 goroutine 的任务发生错误时，会取消所有 goroutine 并返回该错误。
1package main
2
3import (
4 "context"
5 "fmt"
6 "golang.org/x/sync/errgroup"
7)
8
9func main() {
10 g, ctx := errgroup.WithContext(context.Background())
11 _ = ctx
12
13 for i := 0; i < 10; i++ {
14 i := i
15 g.Go(func() error {
16 if i == 5 {
17 return fmt.Errorf("error")
18 }
19 return nil
20 })
21 }
22
23 if err := g.Wait(); err != nil {
24 fmt.Println(err)
25 }
26}
上述代码使用 errgroup 创建 10 个 goroutine，并在第 5 个 goroutine 中引发错误。为了展示错误发生的情况，我们有意在第五个 goroutine 中引发错误。但在实际使用时，应使用 errgroup 创建 goroutine，并针对各 goroutine 中发生错误的情况进行各种后处理。
once
这是一种只应执行一次的代码的执行工具。可以通过下面的构造函数来执行相关代码。
1func OnceFunc(f func()) func()
2func OnceValue[T any](f func() T) func() T
3func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2)
OnceFunc
OnceFunc 仅保证该函数在整个程序中只执行一次。
1package main
2
3import "sync"
4
5func main() {
6 once := sync.OnceFunc(func() {
7 println("Hello, World!")
8 })
9
10 once()
11 once()
12 once()
13 once()
14 once()
15}
上述代码使用 sync.OnceFunc 输出 Hello, World!。在该代码中，使用 sync.OnceFunc 创建 once 函数，即使多次调用 once 函数，Hello, World! 也只会输出一次。
OnceValue
OnceValue 并非仅保证该函数只执行一次，而是会保存该函数的返回值，在再次调用时返回已保存的值。
1package main
2
3import "sync"
4
5func main() {
6 c := 0
7 once := sync.OnceValue(func() int {
8 c += 1
9 return c
10 })
11
12 println(once())
13 println(once())
14 println(once())
15 println(once())
16 println(once())
17}
上述代码使用 sync.OnceValue 使变量 c 每次增加 1。在该代码中，使用 sync.OnceValue 创建 once 函数，即使多次调用 once 函数，变量 c 也只会增加一次，并返回增加后的值 1。
OnceValues
OnceValues 的作用与 OnceValue 相同，但可以返回多个值。
1package main
2
3import "sync"
4
5func main() {
6 c := 0
7 once := sync.OnceValues(func() (int, int) {
8 c += 1
9 return c, c
10 })
11
12 a, b := once()
13 println(a, b)
14 a, b = once()
15 println(a, b)
16 a, b = once()
17 println(a, b)
18 a, b = once()
19 println(a, b)
20 a, b = once()
21 println(a, b)
22}
上述代码使用 sync.OnceValues 使变量 c 每次增加 1。在该代码中，使用 sync.OnceValues 创建 once 函数，即使多次调用 once 函数，变量 c 也只会增加一次，并返回增加后的值 1。
atomic
atomic 包是提供原子操作的包。atomic 包提供 Add、CompareAndSwap、Load、Store、Swap 等方法，但近来更推荐使用 Int64、Uint64、Pointer 等类型。
1package main
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8func main() {
9 wg := sync.WaitGroup{}
10 c := atomic.Int64{}
11
12 for i := 0; i < 100 ; i++ {
13 wg.Add(1)
14 go func() {
15 defer wg.Done()
16 c.Add(1)
17 }()
18 }
19
20 wg.Wait()
21 println(c.Load())
22}
这是之前使用过的示例。该代码使用 atomic.Int64 类型以原子方式增加变量 c。通过 Add 方法和 Load 方法，可以原子地增加变量并读取变量的值。此外，可以使用 Store 方法存储值，使用 Swap 方法交换值，并使用 CompareAndSwap 方法比较值，若符合条件则进行替换。
cond
sync.Cond
cond 包是提供条件变量的包。cond 包可以通过 sync.Cond 创建，并提供 Wait、Signal、Broadcast 方法。
1package main
2
3import (
4 "sync"
5)
6
7func main() {
8 c := sync.NewCond(&sync.Mutex{})
9 ready := false
10
11 go func() {
12 c.L.Lock()
13 ready = true
14 c.Signal()
15 c.L.Unlock()
16 }()
17
18 c.L.Lock()
19 for !ready {
20 c.Wait()
21 }
22 c.L.Unlock()
23
24 println("Ready!")
25}
上述代码使用 sync.Cond 等待变量 ready 变为 true。在该代码中，使用 sync.Cond 等待 ready 变量变为 true 后，输出 Ready!。以这种方式使用 sync.Cond，可以让多个 goroutine 同时等待某个特定条件得到满足。
利用此特性，可以实现一个简单的 queue。
1package queue
2
3import (
4 "sync"
5 "sync/atomic"
6)
7
8type Node[T any] struct {
9 Value T
10 Next *Node[T]
11}
12
13type Queue[T any] struct {
14 sync.Mutex
15 Cond *sync.Cond
16 Head *Node[T]
17 Tail *Node[T]
18 Len int
19}
20
21func New[T any]() *Queue[T] {
22 q := &Queue[T]{}
23 q.Cond = sync.NewCond(&q.Mutex)
24 return q
25}
26
27func (q *Queue[T]) Push(value T) {
28 q.Lock()
29 defer q.Unlock()
30
31 node := &Node[T]{Value: value}
32 if q.Len == 0 {
33 q.Head = node
34 q.Tail = node
35 } else {
36 q.Tail.Next = node
37 q.Tail = node
38 }
39 q.Len++
40 q.Cond.Signal()
41}
42
43func (q *Queue[T]) Pop() T {
44 q.Lock()
45 defer q.Unlock()
46
47 for q.Len == 0 {
48 q.Cond.Wait()
49 }
50
51 node := q.Head
52 q.Head = q.Head.Next
53 q.Len--
54 return node.Value
55}
如此使用 sync.Cond，就不必通过 spin-lock 占用大量 CPU，而是可以高效地等待，并在条件满足时恢复运行。
semaphore
golang.org/x/sync/semaphore.Semaphore
semaphore 包是提供信号量的包。semaphore 包可以通过 golang.org/x/sync/semaphore.Semaphore 创建，并提供 Acquire、Release、TryAcquire 方法。
1package main
2
3import (
4 "context"
5 "fmt"
6 "golang.org/x/sync/semaphore"
7)
8
9func main() {
10 s := semaphore.NewWeighted(1)
11
12 if s.TryAcquire(1) {
13 fmt.Println("Acquired!")
14 } else {
15 fmt.Println("Not Acquired!")
16 }
17
18 s.Release(1)
19}
上述代码使用 semaphore 创建信号量，并通过 Acquire 方法获取信号量，再通过 Release 方法释放信号量。该代码展示了使用 semaphore 获取和释放信号量的方法。
结语
基本内容到这里应该就足够了。希望您能够基于本文的内容理解如何使用 goroutine 管理并发，并能够在实际中加以运用。希望本文对您有所帮助。谢谢。