^-^
【go语言】Golang并发编程
最近更新：2024-12-05   |   字数总计：3.2k   |   阅读估时：15分钟   |   阅读量：次
  1. Go并发编程
    1. WaitGroup与Cond等待一组协程
      1. WaitGroup
      2. Cond作用
    2. Mutex、RWMutex、Map与并发安全
      1. Mutex与RWMutex
      2. Sync.Map
    3. Pool对象重复利用与Once数据懒加载
      1. sync.Pool
      2. sync.Once
    4. atomic通过内存共享数据
      1. sync/atomic
    5. context控制协程退出

Go并发编程

WaitGroup与Cond等待一组协程

WaitGroup

  1. 等待一组协程完成
  2. 工作原理：通过计数器来获取协程的完成情况
  3. 启动一个协程时计数器+1，协程退出时计数器-1
  4. 通过wait方法阻塞主协程，等待计数器清零后才能继续执行后续操作

  • 应用场景

    1. 通过协程并执行一组任务，且任务全部完成后才能进行下一步操作的情况
    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
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
package p1

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

func WaitGroupCase(){
	var a, b = 1000, 10000
	start := time.Now()
	for i:=0;i < 10000000000;i++{ // 串行执行
		multi(a, b)
	}
	t := time.Since(start)
	fmt.Println(t)

	start = time.Now()
	wg := sync.WaitGroup {}
	for i := 0 ;i < 20 ;i++{ // 并发
		wg.Add(1) // 每次waitgroup+1
		go func(){
			defer wg.Done() // 结束后waitgroup-1，以闭包的方式引用
			for j := 0 ; j < 500000000 ; j++{
				multi(a, b)
			}
		}()
	}
	wg.Wait() // 等待全部清零
	t = time.Since(start)
	fmt.Println(t)
}

func multi(a, b int) int {
	return a * b
}


func WaitGroupCase1(){
	ch := make(chan []int, 1000)
	start := time.Now()
	wg2 := sync.WaitGroup{}
	wg2.Add(1)
	go func(){
		defer wg2.Done()
		i := 0
		for item  := range ch {
			fmt.Println(multi(item[0], item[1]))
			i++
		}
		time.Sleep(3 * time.Second)
		fmt.Println("数据处理完成，条数：",i)
	}()

	wg := &sync.WaitGroup{}
	for i:=0;i<2;i++{
		wg.Add(1)
		wg2.Add(1)
		go func(wg1 *sync.WaitGroup){
			defer wg1.Done()
			defer wg2.Done()
			for j := 0;j<500;j++{
				ch <- []int{i, j}
			}
		}(wg)
	}
	wg.Wait()
	close(ch)
	wg2.Wait()
	t := time.Since(start)
	fmt.Println("总处理时间：",t)
}

Cond作用

  1. 设置一组协程根据条件阻塞，可以根据不同的条件阻塞
  2. 可以根据条件唤醒对应的协程

  • 应用场景

    1. 应用于一发多收的场景，即一组协程需要等待某一个协程完成一些前置准备的情况

  • 注意事项

    1. 被叫方必须持有锁
    2. 主叫方可以持有锁，但允许不持有
    3. 尽可能的减少无效唤醒
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
package p1

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

func ConditionCase() {
	list := make([]int, 0)
	list = append(list, 1,2,3,4)
	fmt.Printf("%p\n", list)
	fmt.Printf("%p\n", &list)
	fmt.Println("---------------------------")
	cond := sync.NewCond(&sync.Mutex{})
	go readList(&list, cond)
	go readList(&list, cond)
	go readList(&list, cond)
	time.Sleep(time.Second)
	initList(&list, cond)
	fmt.Println("---------------------------")
	fmt.Printf("%p\n", list)
	fmt.Printf("%p\n", &list) 
}

func initList(list *[]int, c *sync.Cond){
	fmt.Printf("%p\n", list) // 只是为了说明，传递指针方式，指针指向确实是一致的，但是指针本身是值传递的
	fmt.Printf("%p\n", &list)
	(*list)[2] = 10
	fmt.Printf("%p\n", list)
	fmt.Printf("%p\n", &list)
	// 主叫方，可以持有锁，也可以不持有锁
	c.L.Lock()
	defer c.L.Unlock()
	for i := 0 ; i < 10 ;i++{
		*list = append(*list, i)
	}
	// 唤醒所有条件等待的协程
	// c.Broadcast()
	// 只唤起一个
	c.Signal()
}

func readList(list *[]int, c *sync.Cond){
	// 被叫方，必须持有锁
	c.L.Lock()
	defer c.L.Unlock()
	for len(*list) == 0 {
		fmt.Println("readlist wait")
		c.Wait()
	}
	fmt.Println("list 数据为：",*list)
}

type queue struct {
	list []int
	cond *sync.Cond
}

func newQueue() *queue{
	q := &queue {
		list : []int{},
		cond : sync.NewCond(&sync.Mutex{}),
	}
	return q
}

func (q *queue) Put(item int){
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	q.list = append(q.list, item)
	// 通知一个协程去读
	q.cond.Signal()
}

func (q *queue) GetMany(n int) []int{
	q.cond.L.Lock()
	defer q.cond.L.Unlock()
	for len(q.list) < n {
		q.cond.Wait()
	}
	list := q.list[:n]
	q.list = q.list[n:]
	return list
}

func ConditionQueueCase(){
	q := newQueue()
	wg := sync.WaitGroup{}
	for n := 1 ; n<=10 ;n++{
		wg.Add(1)
		go func(n int){
			defer wg.Done()
			list := q.GetMany(n)
			fmt.Printf("%d:%d\n",n,list)
		}(n)
	}
	for i:=0;i<100;i++{
		q.Put(i)
	}

	wg.Wait()
}

Mutex、RWMutex、Map与并发安全

Mutex与RWMutex

  1. 并发场景下，通过锁机制，解决数据竞争的问题
  • 应用场景
    1. 协程安全
    2. 数据竞争
  • 注意事项
    1. 尽量避免使用锁
    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
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
package p1

import (
	"fmt"
	"sync"
)

func MutexCase(){
	// singleRoutine()
	// multipleRoutine()
	// multipleSafeRoutine()
	multipleSafeRoutineByRWMutex()
}

// 单协程操作
func singleRoutine(){
	mp := make(map[string]int, 0)
	list := []string{"A","B","C","D"}
	for i:=0;i<20;i++{
		for _ , item := range list {
			_, ok := mp[item]
			if !ok {
				mp[item] = 0
			}
			mp[item] += 1
		}
	}
	fmt.Println(mp)
}

// 多协程操作，非协程安全
// 不一定能够正常执行
func multipleRoutine(){
	mp := make(map[string]int, 0)
	list := []string{"A","B","C","D"}
	wg := sync.WaitGroup{}
	for i:=0;i<20;i++{
		wg.Add(1)
		go func(){
			defer wg.Done()
			for _ , item := range list {
				_, ok := mp[item]
				if !ok {
					mp[item] = 0
				}
				mp[item] += 1
			}
		}()
	}
	wg.Wait()
	fmt.Println(mp)
}

// 互斥锁协程安全，但是几乎相当于串行执行
func multipleSafeRoutine(){
	type safeMap struct {
		data map[string] int
		sync.Mutex
	}
	mp := safeMap{
		data : make(map[string]int, 0),
		Mutex : sync.Mutex{},
	}

	list := []string{"A","B","C","D"}
	wg := sync.WaitGroup{}
	for i:=0;i<20;i++{
		wg.Add(1)
		go func(){
			defer wg.Done()
			mp.Lock()
			defer mp.Unlock()
			for _ , item := range list {
				_, ok := mp.data[item]
				if !ok {
					mp.data[item] = 0
				}
				mp.data[item] += 1
			}
		}()
	}
	wg.Wait()
	fmt.Println(mp.data)
}


type cache struct {
	data map[string] string
	sync.RWMutex
}

func newCache() *cache {
	return &cache{
		data : make(map[string]string, 0),
		RWMutex: sync.RWMutex{},
	}
}

func (c *cache) Get(key string) string {
	c.RLock()
	defer c.RUnlock()
	value, ok := c.data[key]
	if ok{
		return value
	}
	return ""
}

func (c * cache) Set(key string, value string) {
	c.Lock()
	defer c.Unlock()
	c.data[key] = value
}

// 读写锁，读锁和读锁是不会互斥的，读锁和写锁是互斥的，写锁和写锁之间也是互斥的
func multipleSafeRoutineByRWMutex(){
	c := newCache()
	wg := sync.WaitGroup{}
	wg.Add(1)
	go func(){
		defer wg.Done()
		c.Set("name", "lin")
	}()
	for i:= 0 ;i < 10 ;i++{
		wg.Add(1)
		go func(){
			defer wg.Done()
			fmt.Println(c.Get("name"))
		}()
	}
	wg.Wait()
}

Sync.Map

  1. 一个线程安全集合，内部通过原子访问和锁机制实现结合的线程安全
  • 应用场景
    1. 适合读多写少的应用场景
    2. 在key值以存在的情况下，可以无锁修改器value，比普通map+锁性能更好
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
package p1

import (
	"fmt"
	"sync"
)

func MapCase(){
	mp := sync.Map{}
	// 设置键值对
	mp.Store("name", "nick")
	mp.Store("email", "nick@qq.com")
	// 通过key获取value，如果不存在就返回nil,ok, 否则返回false
	fmt.Println(mp.Load("name"))
	fmt.Println(mp.Load("email"))
	fmt.Println(mp.Load("hobby"))
	// 通过key获取value，如果不存在则设置指定的value并返回
	// ok 为true表示原来key就存在，false表示key不存在，并设置后返回
	fmt.Println(mp.LoadOrStore("hobby", "篮球"))
	fmt.Println(mp.LoadOrStore("email", "nick@qq.com"))
	// 根据key获取value后，删除该key
	// ok 为true表示原来key就存在，false表示key不存在
	fmt.Println(mp.LoadAndDelete("hobby"))
	fmt.Println(mp.LoadAndDelete("hobby"))

	// 为集合设置迭代函数，将为集合中的每一个键值对顺序调用该函数，如果该函数返回false，则停止迭代
	// 为遍历集合中所有元素提供便利
	mp.Range(func(key, value any) bool {
		fmt.Println(key, value)
		return true
	})
}


// 发现结果并不符合预期，所以多个sync.Map的操作语句结合起来并不是一个线程安全的操作，只有单条语句能保证是安全的
func MapCase1() {
	mp := sync.Map{}
	list := []string{"A","B","C","D"}
	wg := sync.WaitGroup{}
	for i:=0;i<20;i++{
		wg.Add(1)
		go func(){
			defer wg.Done()
			for _ , item := range list {
				val := 0
				value, ok := mp.Load(item)
				if !ok {
					val = 0
					value, _ = mp.LoadOrStore(item, 0)
				}
				val = value.(int)
				val += 1
				mp.Store(item, val)
			}
		}()
	}
	wg.Wait()
	mp.Range(func(key, value any) bool {
		fmt.Println(key, value)
		return true
	})
}

Pool对象重复利用与Once数据懒加载

sync.Pool

  1. 创建一个临时对象池，缓存一组对象用于重复利用，以此来减少内存分配和减低GC的压力
  • 应用场景
    1. 可用于连接池（grpc客户端、网络连接等）场景
  • 注意事项
    1. 用户缓存一些创建成本较高，使用比较频繁的对象
    2. Pool的长度默认为机器CPU线程数
    3. 存储在Pool的对象随时都可能在不被通知的情况下被回收
    4. 没有什么创建成本的对象不建议使用对象池
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
package p1

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

func PoolCase(){
	target := "127.0.0.1"
	pool, err := GetPool(target)
	if err != nil {
		log.Fatal(err)
	}
	wg := sync.WaitGroup{}
	for i:= 0 ;i < 10 ;i++{
		wg.Add(1)
		go func(){
			defer wg.Done()
			for j:= 0 ;j < 5 ;j++{
				conn := pool.Get()
				fmt.Println(conn.ID)
				pool.Put(conn)
			}
		}()
	}
	wg.Wait()
}

const (
	ON = 1
	OFF = 0
)

type Conn struct { // 模仿一个连接
	ID int64
	Target string
	Status int
}

func NewConn(target string) *Conn{
	return &Conn{
		ID : rand.Int63(),
		Target : target,
		Status: ON,
	}
}

func (c *Conn) GetStatus () int {
	return c.Status
}

type ConnPool struct {
	sync.Pool
}

func GetPool(target string) (*ConnPool, error) {
	return &ConnPool{
		Pool: sync.Pool {
			New : func () any {
				return NewConn(target)
			},
		},
	}, nil
}

// get
func (c *ConnPool) Get() *Conn {
	conn := c.Pool.Get().(*Conn)
	if conn.GetStatus() == OFF{
		conn = c.Pool.New().(*Conn)
	}
	return conn
}
// put
func (c *ConnPool) Put (conn *Conn){
	if conn.GetStatus() == OFF{
		return 
	}
	c.Pool.Put(conn)
}

sync.Once

  • 用来初始化单例资源
  • 使用场景
    1. 单例场景
    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
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
package p1

import (
	"fmt"
	"sync"
)

type onceMap struct {
	sync.Once
	data map[string] int
}

func (m *onceMap) LoadData(){ // loadData实际上会执行10次
	m.Once.Do(func(){ // 但Do方法只会执行1次
		list := []string{"A","B","C","D"}
		for _,item := range list {
			_, ok := m.data[item]
			if !ok {
				m.data[item] = 0
			}
			m.data[item] += 1
		}
	})
}

// 使用了Once对象，虽然循环10次，但计数只会是1
func OnceCase () {
	o := &onceMap {
		data : make (map[string] int),
	}
	wg := sync.WaitGroup{}
	for i:= 0 ;i < 10 ;i++{
		wg.Add(1)
		go func(){
			defer wg.Done()
			o.LoadData()
		}()
	}
	wg.Wait()
	fmt.Println(o.data)
}

atomic通过内存共享数据

sync/atomic

  1. atomic包提供底层的原子级内存操作，用于实现同步算法
  • 应用场景
    1. 通过内存实现通信
  • 注意事项
    1. atomic属于底层原子操作，如无必要请使用通道或者sync包的其他函数或类型实现同步算法
    2. 如果没有把握正确使用atomic那么请不要使用
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
package p1

import (
	"fmt"
	"sync/atomic"
	"sync"
)

func AtomicCase() {
	var count int64 = 5
	atomic.StoreInt64(&count, 10)
	fmt.Println("获取值：", atomic.LoadInt64(&count))
	// count += 10
	fmt.Println("在原有的基础上累加：", atomic.AddInt64(&count, 10))
	fmt.Println("交换变量的值，并返回原有的值：", atomic.SwapInt64(&count, 30))
	fmt.Println("获取值：", atomic.LoadInt64(&count))

	// 相等替换，不相等不替换
	fmt.Println("比较是否和提供的旧值相等并替换原有的值，并返回是否替换成功：", atomic.CompareAndSwapInt64(&count, 30, 60))
}

type atomicCounter struct {
	count int64
}

func (c *atomicCounter) Inc(){
	atomic.AddInt64(&c.count, 1)
}

func (c *atomicCounter) Load() int64{
	return atomic.LoadInt64(&c.count)
}

func AtomicCase1(){
	var count int64
	wg := sync.WaitGroup{}
	for i:= 0;i<10000;i++{
		wg.Add(1)
		go func(){
			defer wg.Done()
			count += 1 
		}()
	}
	wg.Wait()
	fmt.Println(count)// 协程不安全,并不一定是10000

	// 加锁解法
	var count2 int64
	wg = sync.WaitGroup{}
	locker := sync.Mutex{}
	for i:= 0;i<10000;i++{
		wg.Add(1)
		go func(){
			defer wg.Done()
			locker.Lock()
			count2 += 1 
			locker.Unlock()
		}()
	}
	wg.Wait()
	fmt.Println(count2)// 协程安全

	// 原子操作
	var count3 = atomicCounter{}
	wg = sync.WaitGroup{}
	for i:= 0;i<10000;i++{
		wg.Add(1)
		go func(){
			defer wg.Done()
			count3.Inc()
		}()
	}
	wg.Wait()
	fmt.Println(count3.Load())// 协程安全
}

func AtomicCase2(){
	list := []string {"A", "B", "C", "D"}
	// 定义一个原子值
	var atomicMp atomic.Value
	// 定义一个集合
	mp := map[string] int {}
	atomicMp.Store(&mp)
	wg := sync.WaitGroup{}
	for i:=0;i<100;i++{
		wg.Add(1)
		go func(){
			defer wg.Done()
			atomicLable:
			m := atomicMp.Load().(*map[string]int) // 不能直接在m上操作，因为是引用拷贝，直接操作，多个协程仍然并发写同一个位置，会出错
			m1 := map[string] int{} // 每个协程建立一个自己的新的map
			for k,v := range *m{
				m1[k] = v
			}
			for _,item := range list{
				_,ok:= m1[item]
				if !ok {
					m1[item] = 0
				}
				m1[item] += 1
			}
			swap := atomicMp.CompareAndSwap(m, &m1)
			if !swap {
				// 需要重新执行逻辑
				goto atomicLable
			}
		}()
	}
	wg.Wait()
	fmt.Println(atomicMp.Load())
}

context控制协程退出

  • 是什么
    1. 译为上下文
    2. 用于进程之间信息和信号传递
    3. 用于服务之间信号和信息传递
  • 应用场景
    1. 用于父子协程间取消信息传递
    2. 用于客户端与服务器间的信息传递
    3. 用于设置请求超时时间等
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
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
package p1

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

func ContextCase(){
	ctx := context.Background()
	ctx = context.WithValue(ctx, "desc", "ContextCase")
	ctx, cancel := context.WithTimeout(ctx, time.Second * 2)
	defer cancel()

	data := [][]int{
		{1,2},
		{3,4},
	}
	ch := make(chan []int)
	go calculate(ctx, ch) 
	/*
		主协程起了一个calculate协程，
		calculate协程起了两个sum、multi子协程
	*/
	for i:=0;i<len(data);i++{
		ch <- data[i]
	}
	time.Sleep(10*time.Second)
	/*
		1 + 2 = 3
		1 + 2 = 2
		3 + 4 = 7
		3 + 4 = 12
		sumContext协程退出，context desc：calculate，错误消息：context deadline exceeded
		muliContext协程退出，context desc：calculate，错误消息：context deadline exceeded
		sumContext协程退出，context desc：calculate，错误消息：context deadline exceeded
		calculate协程退出，context desc：ContextCase，错误消息：context deadline exceeded
		muliContext协程退出，context desc：calculate，错误消息：context deadline exceeded
	*/
}

func calculate(ctx context.Context, data <- chan []int){
	for {
		select {
		case item := <- data :
			ctx := context.WithValue(ctx, "desc", "calculate")
			ch := make (chan []int)
			go sumContext(ctx, ch)
			ch <- item

			ch1 := make (chan []int)
			go muliContext(ctx, ch1)
			ch1 <- item

			// fmt.Println(item)
		case <-ctx.Done():
			desc := ctx.Value("desc").(string)
			fmt.Printf("calculate协程退出，context desc：%s，错误消息：%s\n", desc, ctx.Err())
			return 
		}
	}
}



func sumContext(ctx context.Context, data <- chan []int){
	for {
		select {
		case item := <- data :
			a,b := item[0], item[1]
			res := sum(a, b)
			fmt.Printf("%d + %d = %d\n",a,b,res)
		case <-ctx.Done():
			desc := ctx.Value("desc").(string)
			fmt.Printf("sumContext协程退出，context desc：%s，错误消息：%s\n", desc, ctx.Err())
			return 
		}
	}
}

func muliContext(ctx context.Context, data <- chan []int){
	for {
		select {
		case item := <- data :
			a,b := item[0], item[1]
			res := multi(a, b)
			fmt.Printf("%d + %d = %d\n",a,b,res)
		case <-ctx.Done():
			desc := ctx.Value("desc").(string)
			fmt.Printf("muliContext协程退出，context desc：%s，错误消息：%s\n", desc, ctx.Err())
			return 
		}
	}
}


func sum(a,b int) int {
	return a + b
}
func multi(a, b int) int{
	return a * b
}
2024-12-03 该篇文章被 Cleofwine 归为分类: Go云原生

上一篇：【go语言】Golang网络编程 下一篇：【go语言】Golang语言特性
© 2023-2026 Cleofwine / Email / GitHub / 浙ICP备2024136127号
🌙赠我弯弯一枚月，也赠予我晚星⭐ 本站由Hexo驱动｜使用Hexo-theme-A4主题