golang 确保并发安全性

golang并发安全性

在Golang中，并发安全性通常指的是当多个goroutines同时访问同一个数据结构或资源时，能够保证数据的一致性和完整性，避免数据竞争、死锁等问题

并发安全性案例

案例1

创建 count，起1000个goroutines，做一亿次自增运算，代码如下：

func main() {
	wg := sync.WaitGroup{}
	var count = 0
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := 0; i < 10000; i++ {
				count++
			}
		}()
	}
	wg.Wait()
	fmt.Println(count)
}

预计结果：

100000000

执行结果：

➜  test go run main.go 
656188

原因：

count++非原子性操作，在串行场景下，不会出现一致性问题，但是在并发场景下，多个goroutines并发操作同一个变量，就会出现goroutines间count值相互覆盖的情况，导致一致性问题

案例2

通过rpc接口，并行多次数据请求，并把返回结果做整合

func main() {
	wg := sync.WaitGroup{}
	var list []int
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			// rpc请求，结果并进行整合
			list = append(list, rpcCall()...)
		}()
	}
	wg.Wait()
	fmt.Println(len(list))
}

// rpc 调用
func rpcCall() []int {
	var ret []int
	for i := 0; i < 100; i++ {
		ret = append(ret, rand.Intn(10000))
	}
	return ret
}

预计结果：

100000

执行结果：

➜  test go run main.go 
21500

append 方法不是原子操作，并发情况下存在一致性问题

解决办法

• 使用互斥锁（Mutex）：通过使用互斥锁来保护共享资源的访问，一次只允许一个goroutine访问共享资源，从而避免竞争条件
• 使用原子操作（Atomic Operations）：对于简单的读写操作，可以使用原子操作来保证操作的原子性，避免竞争条件
• 使用通道（Channel）：通过使用通道来进行goroutine之间的通信和同步，避免共享资源的直接访问
• 使用并发安全的数据结构，例如 sync.Map等

案例1 优化方案

通过共享锁解决问题：

func main() {
	// 添加互斥锁
	mu := sync.Mutex{}
	wg := sync.WaitGroup{}
	var count = 0
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := 0; i < 10000; i++ {
				// 锁住共享资源
				mu.Lock()
				count++
				mu.Unlock()
			}
		}()
	}
	wg.Wait()
	fmt.Println(count)
}

执行结果：

➜  test go run main.go 
10000000

通过原子操作解决问题：

func main() {
	wg := sync.WaitGroup{}
	var count int32 = 0
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for i := 0; i < 10000; i++ {
				atomic.AddInt32(&count, 1)
			}
		}()
	}
	wg.Wait()
	fmt.Println(count)
}

执行结果：

➜  test go run main.go 
10000000

案例2 优化方案

通过channel + select-case解决问题

func main() {
	// 创建channel
	ch := make(chan []int, 100)
	var list []int
	for i := 0; i < 1000; i++ {
		go func(ch chan []int) {
			// rpc结果写入channel中
			ch <- rpcCall()
		}(ch)
	}

	// 轮询等待每个goroutine的执行结果
	for i := 0; i < 1000; i++ {
		select {
		case v := <-ch:
			list = append(list, v...)
		}
	}
	fmt.Println(len(list))
}

// rpc 调用
func rpcCall() []int {
	var ret []int
	for i := 0; i < 100; i++ {
		ret = append(ret, rand.Intn(10000))
	}
	return ret
}

执行结果：

➜  test go run main.go 
100000