golang笔记18--go并发多线程

介绍

大家都知道go语言近年来越来越火了，其中有一个要点是go语言在并发场景有很高的性能，比如可以通过启动很多个 goroutine 来执行并发任务，通过Channel 类型实现 goroutine 之间的数据交流。当我们想用go实现高并发的时候，我们要了解常见的并发源语，以便于开发的时候做出最优选择。
本文基于较新版本的go1.20.7, 介绍了go并发多线场景常用的源语和方法案例…

核心用法

Mutex

多线程场景下为了解决资源竞争问题，通常会使用互斥锁，限定临界区只能同时由一个线程持有。
在go语言中是通过Mutex来实现的。

案例见：

package main

import (
	"fmt"
	"sync"
)

func noLock() {
	var count = 0
	var wg sync.WaitGroup
	wg.Add(10)
	for i := 0; i < 10; i++ {
		go func() {
			defer wg.Done()
			for j := 0; j < 10000; j++ {
				count++
			}
		}()
	}
	wg.Wait()
	fmt.Printf("count=%v\n", count)
}

func hasLock() {
	var count = 0
	var wg sync.WaitGroup
	wg.Add(10)
	var mu sync.Mutex
	for i := 0; i < 10; i++ {
		go func() {
			defer wg.Done()
			for j := 0; j < 10000; j++ {
				mu.Lock()
				count++
				mu.Unlock()
			}
		}()
	}
	wg.Wait()
	fmt.Printf("count=%v\n", count)
}

func main() {
	fmt.Println("no lock:")
	for i := 0; i < 10; i++ {
		noLock()
	}
	fmt.Println("has lock:")
	for i := 0; i < 10; i++ {
		hasLock()
	}
}
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

输出：

no lock:
count=53430
count=42448
count=47531
count=57758
count=50497
count=44185
count=41547
count=33113
count=35673
count=31391
has lock:
count=100000
count=100000
count=100000
count=100000
count=100000
count=100000
count=100000
count=100000
count=100000
count=100000
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

很多时候无法快速看出程序是否有竞争问题，此时可以使用race来检查是否有竞争关系

$ go run -race 4.1.go
1

常见错误：

• 不成对出现
• copy已经使用的mutex
  使用 go vet 检查
• 重入
• 死锁

RWMutex

当有大量读写操作的时候，若仅仅使用Mutex会影响性能，此时可以使用读写锁来将读写区分开来；goroutine A持有读锁的时候，其它goroutine也可以继续读操作，写操作goroutine A持有锁的时候，它就是一个排它锁，其它读写操作会阻塞等待锁被释放。
RWMutex是一个reader/writer互斥锁，某一时刻能由任意的reader持有，或只能被单个writer持有。

适用于读多、写少的场景。

案例见：

package main

import (
	"fmt"
	"sync"
	"time"
)

// Counter 线程安全的计数器
type Counter struct {
	mu    sync.RWMutex
	count uint64
}

func (c *Counter) Incr() {
	c.mu.Lock()
	c.count++
	c.mu.Unlock()
}

func (c *Counter) Count() uint64 {
	c.mu.RLock()
	defer c.mu.RUnlock()
	return c.count
}

func main() {
	var count Counter
	for i := 0; i < 10; i++ {
		go func(i int) {
			for {
				ret := count.Count()
				fmt.Printf("reader %v, count=%v\n", i, ret)
				time.Sleep(time.Second * 2)
			}
		}(i)
	}

	for {
		count.Incr()
		fmt.Printf("writer, count=%v\n", count.count)
		time.Sleep(time.Second * 5)
	}
}
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

输出：

writer, count=1
reader 3, count=1
reader 1, count=1
reader 2, count=1
...
reader 0, count=1
reader 3, count=1
reader 5, count=1
reader 4, count=1
reader 9, count=1
reader 7, count=1
writer, count=2
reader 3, count=2
reader 7, count=2
reader 8, count=2
...
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

WaitGroup

WaitGroup就是package sync用来做任务编排的一个并发原语。它要解决的就是并发-等待的问题：现在有一个goroutine A 在检查点（checkpoint）等待一组goroutine全部完成，如果在执行任务的这些goroutine还没全部完成，那么goroutine A就会阻塞在检查点，直到所有goroutine都完成后才能继续执行。

它有如下三个方法：

• Add，用来设置WaitGroup的计数值；
• Done，用来将WaitGroup的计数值减1，其实就是调用了Add(-1)；
• Wait，调用这个方法的goroutine会一直阻塞，直到WaitGroup的计数值变为0。

案例见：

package main

import (
	"fmt"
	"sync"
	"time"
)

type Counter struct {
	mu    sync.Mutex
	count uint64
}

// Incr 计数器加1
func (c *Counter) Incr() {
	c.mu.Lock()
	c.count++
	c.mu.Unlock()
}

// Count 获取count值
func (c *Counter) Count() uint64 {
	c.mu.Lock()
	defer c.mu.Unlock()
	return c.count
}

// worker sleep 1s后加1
func worker(c *Counter, w *sync.WaitGroup, i int) {
	defer w.Done()
	time.Sleep(time.Second)
	c.Incr()
	fmt.Printf("worker %v add 1\n", i)
}

func main() {
	var counter Counter
	var wg sync.WaitGroup
	wg.Add(10)
	for i := 0; i < 10; i++ {
		go worker(&counter, &wg, i)
	}
	wg.Wait()
	fmt.Printf("finished, count=%v\n", counter.count)
}
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

案例中10个worker分别对count+1， 10个worker完成后才输出最终的count。

输出：

worker 8 add 1
worker 6 add 1
worker 3 add 1
worker 1 add 1
worker 2 add 1
worker 4 add 1
worker 5 add 1
worker 7 add 1
worker 9 add 1
worker 0 add 1
finished, count=10
1
2
3
4
5
6
7
8
9
10
11

Cond

Go 标准库提供 Cond 原语的目的是，为等待 / 通知场景下的并发问题提供支持。Cond 通常应用于等待某个条件的一组 goroutine，等条件变为 true 的时候，其中一个 goroutine 或者所有的 goroutine 都会被唤醒执行

案例见：

package main

import (
	"log"
	"math/rand"
	"sync"
	"time"
)

func main() {
	c := sync.NewCond(&sync.Mutex{})
	var ready int
	for i := 0; i < 10; i++ {
		go func(i int) {
			time.Sleep(time.Duration(rand.Int63n(10)) * time.Second)
			c.L.Lock()
			ready++
			c.L.Unlock()
			log.Printf("运动员 %v 已经就绪", i)
			c.Broadcast()
		}(i)
	}

	c.L.Lock()
	for ready != 10 {
		c.Wait()
		log.Printf("裁判员被唤醒一次")
	}
	c.L.Unlock()
	log.Printf("所有运动员就绪，开始比赛 3,2,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

输出：

2023/10/11 23:52:41 运动员 4 已经就绪
2023/10/11 23:52:41 裁判员被唤醒一次
2023/10/11 23:52:43 运动员 7 已经就绪
2023/10/11 23:52:43 裁判员被唤醒一次
2023/10/11 23:52:43 运动员 5 已经就绪
2023/10/11 23:52:43 裁判员被唤醒一次
2023/10/11 23:52:44 运动员 6 已经就绪
2023/10/11 23:52:44 裁判员被唤醒一次
2023/10/11 23:52:44 运动员 3 已经就绪
2023/10/11 23:52:44 裁判员被唤醒一次
2023/10/11 23:52:45 运动员 8 已经就绪
2023/10/11 23:52:45 裁判员被唤醒一次
2023/10/11 23:52:47 运动员 0 已经就绪
2023/10/11 23:52:47 裁判员被唤醒一次
2023/10/11 23:52:48 运动员 1 已经就绪
2023/10/11 23:52:48 裁判员被唤醒一次
2023/10/11 23:52:49 运动员 9 已经就绪
2023/10/11 23:52:49 裁判员被唤醒一次
2023/10/11 23:52:49 运动员 2 已经就绪
2023/10/11 23:52:49 裁判员被唤醒一次
2023/10/11 23:52:49 所有运动员就绪，开始比赛 3,2,1...
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

Once

Once 可以用来执行且仅仅执行一次动作，常常用来初始化单例资源，或者并发访问只需初始化一次的共享资源，或者在测试的时候初始化一次测试资源。

sync.Once 只暴露了一个方法 Do，你可以多次调用 Do 方法，但是只有第一次调用 Do 方法时 f 参数才会执行，这里的 f 是一个无参数无返回值的函数。

func (o *Once) Do(f func())
1

案例见：

package main

import (
	"fmt"
	"net"
	"runtime"
	"sync"
	"time"
)

func runFuncName() string {
	pc := make([]uintptr, 1)
	runtime.Callers(2, pc)
	f := runtime.FuncForPC(pc[0])
	return f.Name()
}
func onceCase1() {
	fmt.Printf("this is %v \n", runFuncName())
	var once sync.Once
	f1 := func() {
		fmt.Println("this is f1")
	}
	f2 := func() {
		fmt.Println("this is f2")
	}
	once.Do(f1)
	once.Do(f2)
}

var conn net.Conn
var once sync.Once

func onceGetConn() net.Conn {
	fmt.Printf("this is %v \n", runFuncName())
	addr := "baidu.com"
	once.Do(func() {
		fmt.Println("this is once.Do")
		conn, _ = net.DialTimeout("tcp", addr+":80", time.Second*10)
	})
	if conn != nil {
		return conn
	} else {
		return nil
	}
}

func main() {
	onceCase1()
	conn = onceGetConn()
	conn = onceGetConn()
	fmt.Println("conn=", conn)
}
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

onceCase1 中可以看到once.Do 中的函数只执行第一次；

onceGetConn 中可以看到单例函数只执行一次初始化；

输出：

this is main.onceCase1 
this is f1
this is main.onceGetConn 
this is once.Do
this is main.onceGetConn 
conn= &{{0xc0000a6180}}
1
2
3
4
5
6

map

Go 内建的 map 对象不是线程（goroutine）安全的，并发读写的时候运行时会有检查，遇到并发问题就会导致 panic。

案例1：

package main

func main() {
	var m = make(map[int]int, 10)
	go func() {
		for {
			m[1] = 1
		}
	}()
	go func() {
		for {
			_ = m[2]
		}
	}()
	select {}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

输出：

fatal error: concurrent map read and map write

goroutine 6 [running]:
main.main.func2()
        /home/xg/files/code/1cc/study/zhangxing12/go/src/chapter04/4.6.go:12 +0x2e
created by main.main
        /home/xg/files/code/1cc/study/zhangxing12/go/src/chapter04/4.6.go:10 +0x8a

goroutine 1 [select (no cases)]:
main.main()
        /home/xg/files/code/1cc/study/zhangxing12/go/src/chapter04/4.6.go:15 +0x8f

goroutine 5 [runnable]:
main.main.func1()
        /home/xg/files/code/1cc/study/zhangxing12/go/src/chapter04/4.6.go:7 +0x2e
created by main.main
        /home/xg/files/code/1cc/study/zhangxing12/go/src/chapter04/4.6.go:5 +0x5d

Process finished with the exit code 2
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

为解决该问题，可以重写线程安全的map，使用第三方的发片式map，或者使用Go 官方线程安全 map 的标准实现 sync.Map, 其使用场景：

• 只会增长的缓存系统中，一个 key 只写入一次而被读很多次；
• 多个 goroutine 为不相交的键集读、写和重写键值对。

案例2：
使用sync.Map 后，并发读写正常

package main

import (
  "fmt"
  "sync"
)

func main() {
  m := sync.Map{}
  go func() {
    for {
      for i := 0; i < 10; i++ {
        m.Store(i, i*10)
      }
    }
  }()
  go func() {
    for {
      v, _ := m.Load(2)
      fmt.Println(v)
    }
  }()
  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

输出：

20
20
20
...
1
2
3
4

Pool

Go 的自动垃圾回收机制还是有一个 STW（stop-the-world，程序暂停）的时间，而且，大量地创建在堆上的对象，也会影响垃圾回收标记的时间。

Go 标准库中提供了一个通用的 Pool 数据结构，也就是 sync.Pool，我们使用它可以创建池化的对象。这个类型也有一些使用起来不太方便的地方，就是它池化的对象可能会被垃圾回收掉，这对于数据库长连接等场景是不合适的。

sync.Pool 本身就是线程安全的，多个 goroutine 可以并发地调用它的方法存取对象；

sync.Pool 不可在使用之后再复制使用。

案例见：

package main

import (
	"bytes"
	"fmt"
	"io"
	"math/rand"
	"os"
	"sync"
	"time"
)

var bufPool = sync.Pool{
	New: func() any {
		return new(bytes.Buffer)
	},
}

func Log(w io.Writer, key, val string) {
	b := bufPool.Get().(*bytes.Buffer)
	b.Reset()
	b.WriteString(time.Now().Local().Format(time.RFC3339))
	b.WriteByte(' ')
	b.WriteString(key)
	b.WriteByte('=')
	b.WriteString(val)
	b.WriteByte('\n')
	w.Write(b.Bytes())
	bufPool.Put(b)
}

func main() {
	rand.New(rand.NewSource(time.Now().UnixNano()))
	for i := 0; i < 10; i++ {
		time.Sleep(time.Second)
		valStr := fmt.Sprintf("/search?=q=flowers %v", rand.Int63n(100))
		Log(os.Stdout, "path", valStr)
	}
}
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

输出：

2023-10-16T14:16:15+08:00 path=/search?=q=flowers 71
2023-10-16T14:16:16+08:00 path=/search?=q=flowers 51
2023-10-16T14:16:17+08:00 path=/search?=q=flowers 21
2023-10-16T14:16:18+08:00 path=/search?=q=flowers 14
2023-10-16T14:16:19+08:00 path=/search?=q=flowers 42
2023-10-16T14:16:20+08:00 path=/search?=q=flowers 15
2023-10-16T14:16:21+08:00 path=/search?=q=flowers 19
2023-10-16T14:16:22+08:00 path=/search?=q=flowers 53
2023-10-16T14:16:23+08:00 path=/search?=q=flowers 45
2023-10-16T14:16:24+08:00 path=/search?=q=flowers 60
1
2
3
4
5
6
7
8
9
10

Context

Go 标准库的 Context 不仅提供了上下文传递的信息，还提供了 cancel、timeout 等其它信息; 它比较适合使用在如下场景：

• 上下文信息传递 （request-scoped），比如处理 http 请求、在请求处理链路上传递信息；
• 控制子 goroutine 的运行；
• 超时控制的方法调用；
• 可以取消的方法调用。

案例见：

package main

import (
	"context"
	"fmt"
	"runtime"
	"time"
)

var neverReady = make(chan struct{})

const shortDuration = 1 * time.Microsecond

func runFuncName() string {
	pc := make([]uintptr, 1)
	runtime.Callers(2, pc)
	f := runtime.FuncForPC(pc[0])
	return f.Name()
}

func case1WithCancel() {
	fmt.Printf("this is %v\n", runFuncName())
	gen := func(ctx context.Context) <-chan int {
		dst := make(chan int)
		n := 1
		go func() {
			for {
				select {
				case <-ctx.Done():
					return // returning not to leak the goroutine
				case dst <- n:
					n++
				}
			}
		}()
		return dst
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel() // cancel when we are finished consuming integers

	for n := range gen(ctx) {
		fmt.Println(n)
		if n == 5 {
			break
		}
	}
}

func case2WithDeadline() {
	fmt.Printf("this is %v\n", runFuncName())
	d := time.Now().Add(shortDuration)
	ctx, cancel := context.WithDeadline(context.Background(), d)
	defer cancel()

	select {
	case <-neverReady:
		fmt.Println("ready")
	case <-time.After(2 * time.Second):
		fmt.Printf("overslept %v\n", 2*time.Second)
	case <-ctx.Done():
		fmt.Println("ctx.Done:", ctx.Err())
	}
}

func case3WithTimeout() {
	ctx, cancel := context.WithTimeout(context.Background(), shortDuration)
	defer cancel()
	select {
	case <-neverReady:
		fmt.Println("ready")
	case <-ctx.Done():
		fmt.Println("ctx.Done:", ctx.Err())
	}
}

func main() {
	fmt.Println(time.Now().Local())
	case1WithCancel()
	fmt.Println(time.Now().Local())
	case2WithDeadline()
	fmt.Println(time.Now().Local())
	case3WithTimeout()
	fmt.Println(time.Now().Local())
}
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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

输出：

2023-10-16 16:41:32.05194173 +0800 CST
this is main.case1WithCancel
1
2
3
4
5
2023-10-16 16:41:32.052263636 +0800 CST
this is main.case2WithDeadline
ctx.Done: context deadline exceeded
2023-10-16 16:41:32.052326891 +0800 CST
ctx.Done: context deadline exceeded
2023-10-16 16:41:32.052351282 +0800 CST
1
2
3
4
5
6
7
8
9
10
11
12
13

select

select 是 Go 中的一个控制结构，类似于 switch 语句。
select 语句只能用于通道操作，每个 case 必须是一个通道操作，要么是发送要么是接收。
select 语句会监听所有指定的通道上的操作，一旦其中一个通道准备好就会执行相应的代码块。
如果多个通道都准备好，那么 select 语句会随机选择一个通道执行。如果所有通道都没有准备好，那么执行 default 块中的代码。

Go 编程语言中 select 语句的语法如下：

select {
case <- channel1:
// 执行的代码
case value := <- channel2:
// 执行的代码
case channel3 <- value:
// 执行的代码
// 你可以定义任意数量的 case
default:
// 所有通道都没有准备好，执行的代码
}
1
2
3
4
5
6
7
8
9
10
11

上述语法中：
每个 case 都必须是一个通道
所有 channel 表达式都会被求值
所有被发送的表达式都会被求值
如果任意某个通道可以进行，它就执行，其他被忽略。
如果有多个 case 都可以运行，select 会随机公平地选出一个执行，其他不会执行。
否则：

• 如果有 default 子句，则执行该语句。
• 如果没有 default 子句，select 将阻塞，直到某个通道可以运行；Go 不会重新对 channel 或值进行求值

如下，两个 goroutine 定期分别输出 one two到通道c1 c2中， 通过 select 来接受数据

package main

import (
  "fmt"
  "time"
)

func main() {

  c1 := make(chan string)
  c2 := make(chan string)

  go func() {
    for {
      time.Sleep(1 * time.Second)
      c1 <- fmt.Sprint("one", time.Now().Local())
    }
  }()
  go func() {
    for {
      time.Sleep(2 * time.Second)
      c2 <- fmt.Sprint("two", time.Now().Local())
    }
  }()

  for {
    select {
    case msg1 := <-c1:
      fmt.Println("received", msg1)
    case msg2 := <-c2:
      fmt.Println("received", msg2)
    }
  }
}
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

输出：

received one2023-10-16 17:27:50.605975411 +0800 CST
received two2023-10-16 17:27:51.606263901 +0800 CST
received one2023-10-16 17:27:51.607610553 +0800 CST
received one2023-10-16 17:27:52.608383998 +0800 CST
received two2023-10-16 17:27:53.606825344 +0800 CST
received one2023-10-16 17:27:53.609350218 +0800 CST
...
1
2
3
4
5
6
7

注意golang中select为空的话会导致语法检测为死锁，因此要禁止如下写法
案例见： src/chapter04/4.9.go

package main

import "fmt"

func foo() {
	fmt.Printf("hi this is foo\n")
}

func bar() {
	fmt.Printf("hi this is bar\n")
}

func main() {
	go foo()
	go bar()
	select {}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

输出：

hi this is bar
hi this is foo
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan receive]:
main.main()
        /home/xg/files/code/1cc/study/zhangxing12/go/src/chapter04/4.9.go:19 +0x3f
1
2
3
4
5
6
7

go 中select为了避免饥饿会随机执行case， 具体见如下案例：
案例中既不全Receive C也不全 Receive S

package main

import (
	"fmt"
	"time"
)

func genInt(ch chan int, stopCh chan bool) {
	for j := 0; j < 10; j++ {
		ch <- j
		time.Sleep(time.Second)
	}
	stopCh <- true
}

func main() {

	ch := make(chan int)
	c := 0
	stopCh := make(chan bool)

	go genInt(ch, stopCh)

	for {
		select {
		case c = <-ch:
			fmt.Println("Receive C", c)
		case s := <-ch:
			fmt.Println("Receive S", s)
		case _ = <-stopCh:
			goto end
		}
	}
end:
}
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

输出：

Receive C 0
Receive S 1
Receive C 2
Receive S 3
Receive C 4
Receive C 5
Receive S 6
Receive S 7
Receive S 8
Receive S 9
1
2
3
4
5
6
7
8
9
10

注意事项

待补充

参考文档

深度解密Go语言之sync.map
go 并发变成实战课
Go 语言 select 语句
Concurrency in Go
Go 语言条件语句