几段 Go 并发代码

Posted 2022-01-09 机智的小小帅

几段 Go 并发代码

可以使用 go run --race main.go 来验证代码中是否存在并发问题

for range

for i,v := range slice 
  // ...
  // go func() ...

在 for range 中，i, v 这两个变量仅仅被初始化一次，在之后在每轮 for 循环中都会修改 i 和 v 的值，所以如果在 for 循环里面开启一个 goroutine 引用 i 和 v 的话，由于 for range 所在的协程在对 i, v 进行写操作, 每轮 for 循环的内部的协程对 i, v 进行读操作，很容易导致并发问题。

解决方法是，在 for 循环内部声明个临时变量，将 i, v 的值赋值给该临时变量，而循环内部的协程就不再直接引用 i, v 了，而是引用该临时变量。

第一个版本

type T struct 
   FieldA string


func main() 
	const elementCount = 100
	sliceT := make([]*T, elementCount, elementCount)
	for i := range sliceT 
		sliceT[i] = new(T)
	

	wg := sync.WaitGroup
	wg.Add(elementCount)

	for _, t := range sliceT 
		go func() 
			defer wg.Done()
			t.FieldA = "hello"
		()
	
	wg.Wait()

修复后

type T struct 
	FieldA string


func main() 
	const elementCount = 100
	sliceT := make([]*T, elementCount, elementCount)
	for i := range sliceT 
		sliceT[i] = new(T)
	

	wg := sync.WaitGroup
	wg.Add(elementCount)

	for _, t := range sliceT 
		localT := t
		go func() 
			defer wg.Done()
			localT.FieldA = "hello"
		()
	

	wg.Wait()

第二个版本

type T struct 
	FieldA string


func main() 
	const elementCount = 100
	sliceT := make([]*T, elementCount, elementCount)
	for i := range sliceT 
		sliceT[i] = new(T)
	

	wg := sync.WaitGroup
	wg.Add(elementCount)

	for i := range sliceT 
		go func() 
			defer wg.Done()
			sliceT[i].FieldA = "hello"
		()
	

	wg.Wait()

修复后

type T struct 
	FieldA string


func main() 
	const elementCount = 100
	sliceT := make([]*T, elementCount, elementCount)
	for i := range sliceT 
		sliceT[i] = new(T)
	

	wg := sync.WaitGroup
	wg.Add(elementCount)

	for i := range sliceT 
		localI := i
		go func() 
			defer wg.Done()
			sliceT[localI].FieldA = "hello"
		()
	

	wg.Wait()

sync.WaitGroup

WaitGroup 封装了一个计数器并提供了三个 api ，分别是 Add(), Done(), Wait()。

Add(n int) 将会给计数器增加 n。

Done() 将会给计数器减 1。

Wait() 将会一直阻塞，直到计数器的值变成 0。

下面是一个常见错误:

type T struct 
	FieldA string


func main() 
	const elementCount = 100
	sliceT := make([]*T, elementCount, elementCount)
	for i := range sliceT 
		sliceT[i] = new(T)
	

	wg := sync.WaitGroup

	for i := range sliceT 
		localI := i
		go func() 
			wg.Add(1)
			defer wg.Done()
			sliceT[localI].FieldA = "hello"
		()
	
	
	wg.Wait()

这段程序不能保证 go func() 里面的代码一定会被执行。

因为 wg.Add(1) 是在子协程里面做的，程序完全可以执行完 for 循环之后，在 wg.Wait() 直接返回，不给子协程执行的机会。而由于 wg 里面的计数器为 0，wg.Wait() 并不会被阻塞。

由此得到一个很重要的结论: wg.Add() 必须和 wg.Wait() 在同一个协程里面被调用。

封装 WaitGroup

有时我们并不想显示地调用 wg.Add() 。我们仅仅想并发执行几段代码，并用 wg.Wait() 等待这几段代码全部执行完成。

看看下面的程序

type T struct 
	FieldA string


func main() 
	const elementCount = 100
	sliceT := make([]*T, elementCount, elementCount)
	for i := range sliceT 
		sliceT[i] = new(T)
	
	
	wg := sync.WaitGroup
	wgFunc := func(do func()) 
		wg.Add(1)
		defer wg.Done()
		do()
	

  for i := range sliceT 
		localI := i
    go wgFunc(func() 
			sliceT[localI].FieldA = "hello"
		)
	

	wg.Wait()

wgFunc 方法内部引用了 wg，每次调用时会执行 wg.Add(1) 和 defer wg.Done() 来为并发保价护航。在 for 循环内部每次会开启一个协程去调用 wgFunc()。

但是它犯了和前一段代码一样的错误。

正确的封装应该是：

func main() 
	const elementCount = 100
	sliceT := make([]*T, elementCount, elementCount)
	for i := range sliceT 
		sliceT[i] = new(T)
	

	wg := sync.WaitGroup
	wgFunc := func(do func()) 
		wg.Add(1)
		go func() 
			defer wg.Done()
			do()
		()
	

	for i := range sliceT 
		localI := i
		wgFunc(func() 
			sliceT[localI].FieldA = "hello"
		)
	

	wg.Wait()

再重复一次，wg.Add() 必须和 wg.Wait() 在同一个协程内被执行。

PS: golang.org/x/sync/errgroup 提供了比 sync.WaitGroup 功能更加强大的 errgroup.Group，用它，用它！

多个协程并发读写同一个变量时一定需要使用同步机制么？

不一定。

stack overflow 上面的一个问题: Can I concurrently write different slice elements

The rule is simple: if multiple goroutines access a variable concurrently, and at least one of the accesses is a write, then synchronization is required.

Structured variables of array, slice, and struct types have elements and fields that may be addressed individually. Each such element acts like a variable.

数组里面的每个元素/结构体里面的每个字段是能被独立寻址的，它们都可以视为一个独立的变量！

因此上面的程序虽然有很多协程在访问同一个 slice，但是由于这些协程访问的是 slice 上不同的元素。且 slice 上的每个元素只被一个协程访问，因此是安全的。

同理，多个协程访问结构体的不同字段也是安全的。

type T struct 
	FieldA string
	FieldB string


func main() 
	sliceT := make([]*T, 1000)
	for i := range sliceT 
		sliceT[i] = new(T)
	

	wg := sync.WaitGroup
	wgFunc := func(do func()) 
		wg.Add(1)
		go func() 
			defer wg.Done()
			do()
		()
	

	for i := range sliceT 
		t := sliceT[i]
		wgFunc(func() 
			t.FieldA = "hello"
		)
		wgFunc(func() 
			t.FieldB = "world"
		)
	

	wg.Wait()