QingZhi's Blog

Persistence is not easy

0%

Go Concurrency programming Sync

Posted on In Views: Valine:

sync包是Go标准包之一, 其使用场景可以分为三种:

  1. Mutex: 共享锁, 读写锁用于共享资源的互斥使用;
  2. WaitGroup: goroutine同步控制;
  3. Once: 全局唯一初始化代码.

在并发编程过程中, 涉及到前两种使用场景. 本文会简单采用Demo方式介绍Mutex, WaitGroup的使用.

Demo

Step1:

首先我们将定义一个并发不安全的计数器, 在单协程的场景下运行, 并检验其结果, 示例代码如下:

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
package step1

import (
	"fmt"
	"time"
)

type Counter struct {
	Count int
}

func (c *Counter) Inc() {
	time.Sleep(10 * time.Millisecond)
	c.Count++
}

func (c *Counter) Value() int {
	return c.Count
}

func assertCount(c *Counter, want int) {
	if c.Value() == want {
		fmt.Println("Assert SUCCESS!!!")
	} else {
		fmt.Printf("Assert FAILED, want: %v, got: %v \n", want, c.Value())
	}
}

func Assert() {
	c := &Counter{}
	for i := 0; i < 10; i++ {
		c.Inc()
	}

	assertCount(c, 10)
}

输出为: Assert SUCCESS!!!

上面的代码中首先定义了一个计数器Counter, 其存在两个方法Inc:计数自增方法, Value:获得计数方法, 这是一种并发不安全的定义写法. 其次定义了一个assertCount工具类函数, 用于检验计数器计数是否与预期一致;最后在Assert函数, 通过for loop的方式调用了10次自增方法, 随后检验是否与预期计数一致. 结果是预期一致的;

Step2:

单协程运行自然是安全而稳定的, 但是对于性能表现必然是不及多并发环境的, 为了验证并发前后的性能差距, 我这边编写了benchmarkTest:

1
2
3
4
5
func BenchmarkAssert(b *testing.B) {
	for i := 0; i < b.N; i++ {
		Assert()
	}
}

一次执行结果如下:

1
2
3
4
5
6
7
8
go test -bench .

Assert SUCCESS!!!
.... // 10次 Assert SUCCESS!!!
Assert SUCCESS!!!
10         101500306 ns/op
PASS
ok      learngo/pkg/sync/cp/step1       1.230s

执行次数10次, 每次时间开销为101500306 ns, 总时间开销为1.230s;

添加sync并发控制代码: 启动10个goroutine去执行:调整Assert函数如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
func Assert() {
	c := &Counter{}
	var wg sync.WaitGroup

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func() {
			c.Inc()
			wg.Done()
		}()
	}
	wg.Wait()
	assertCount(c, 10)
}

在上面的示例函数中, 我们利用sync.WaitGroup实现了goroutines的同步控制:

1
2
3
4
var wg sync.WaitGroup // 声明一个waitgroup变量, 声明之后即可使用, 无需特殊初始化
wg.Add(1) // 协程计数器+1, 通常与协程创建同步发生
wg.Done() // 协程计数器-1, 协程内操作完成后执行
wg.Wait() // wg阻塞操作, 直到计数器==0

再次执行benchmark:

1
2
3
4
5
go test -bench .
....
100          10211660 ns/op
PASS
ok      learngo/pkg/sync/cp/step2       1.046s

发现: 执行次数为100次, 每次执行时间开销为10211660 ns纳秒, 总时间开销为1.046s.发现性能表现显著提升.但是, 如果将Assert函数中的计数次数提升, 会发现其并发并不安全: 例如我将协程更改为1000次, 预测结果为1000的话:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
func Assert1000() {
	c := &Counter{}
	var wg sync.WaitGroup

	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			c.Inc()
			wg.Done()
		}()
	}
	wg.Wait()
	assertCount(c, 1000)
}

func TestAssert(t *testing.T) {
	Assert1000()
}
1
2
3
4
5
6
go test  -v .
=== RUN   TestAssert
Assert FAILED, want: 1000, got: 983 
--- PASS: TestAssert (0.01s)
PASS
ok      learngo/pkg/sync/cp/step2       0.015s

这是由于多个协程对于共享资源counter.Count的访问时并发不安全的, 需要借助于sync.Mutex / sync.RWMutex, 进行资源访问并发控制;

Step3:

sync.Mutex 是互斥锁, 适用于写多读少场景. sync.RWMutex是读写锁, 适用于读多写少的场景;本用例的场景为写多, 所以可以采用sync.Mutex, 调整后代码如下:

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
package step3

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

type Counter struct {
	mu    sync.Mutex
	Count int
}

func (c *Counter) Inc() {
	time.Sleep(10 * time.Millisecond)
	c.mu.Lock()
	defer c.mu.Unlock()
	c.Count++
}

func (c *Counter) Value() int {
	return c.Count
}

...

我们在Counter中添加了属性mu sync.Mutex, 用于控制Counter的同步访问. 在访问Counter前, 通过c.mu.Lock()加锁, 访问结束后通过c.mu.Unlock()解锁;再次执行测试函数:

1
2
3
4
5
6
go test  -v .
=== RUN   TestAssert
Assert SUCCESS!!!
--- PASS: TestAssert (0.01s)
PASS
ok      learngo/pkg/sync/cp/step3       (cached)

发现执行通过.

需要注意的是:

  1. sync.Mutex在声明后是不允许复制的, 这是因为值传递的效果导致出现了另一把锁, 会使得预期不符.

  2. unbuffered channel也可以做为所出现, 那么什么情况下需要使用mutex? 

    1
    2
    3
    Paraphrasing:
    • Use channels when passing ownership of data
    • Use mutexes for managing state
    • 当传递的是数据的所有权时建议使用channel
    • 当传递的时数据的状态时建议使用mutex

结论

本文总结了在并发编程中sync的使用场景, 一共分为两种:

  1. sync.Waitgroup用于goroutine同步控制;
  2. sync.Mutex用于资源的互斥访问.

并通过Demo的方式一步一步的展示了两种使用方式, 理解起来应该是简单的;

在实际项目中, channel, sync在很多场景的作用时相同的, 既可以使用channel,也可以使用sync,甚至会出现混用的情况.个人认为不需要纠结, 只需要编写符合项目风格, 可读性更好的代码即可;