前言

在学习操作系统的时候,我们应该都学习过临界区、互斥锁这些概念,用于在并发环境下保证状态的正确性。比如在秒杀时,100 个用户同时抢 10 个电脑,为了避免少卖或者超卖,就需要使用锁来进行并发控制。 在 Go语言 里面互斥锁是 sync.Mutex ,我们本篇文章就来学习下为什么要使用互斥锁、如何使用互斥锁,以及使用时的常见问题。

为什么要使用互斥锁

我们来看一个示例:我们起了 10000 个协程将变量 num 加1,因此肯定会存在并发,如果我们不控制并发,10000 个协程都执行完后,该变量的值很大概率不等于 10000。

那么为什么会出现这个问题呢,原因是 num++ 不是原子操作,它会先读取变量 num 当前值,然后对这个值 加1,再把结果保存到 num 中。例如 10goroutine 同时运行到 num++ 这一行,可能同时读取 num=1000,都加1后再保存, num=1001,这就与想要的结果不符。

 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 main

import (
	"fmt"
	"sync"
)

func main() {
	num := 0

	var wg sync.WaitGroup
	threadCount := 10000
	wg.Add(threadCount)
  
	for i := 0; i < threadCount; i++ {
		go func() {
			defer wg.Done()
			num++
		}()
	}
  
	wg.Wait() // 等待 10000 个协程都执行完
  fmt.Println(num) // 9388(每次都可能不一样)

}

我们如果使用了互斥锁,可以保证每次进入临界区的只有一个 goroutine,一个 goroutine 执行完后,另一个 goroutine 才能进入临界区执行,最终就实现了并发控制。

并发获取锁示意图

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

import (
	"fmt"
	"sync"
)

func main() {
	num := 0
	var mutex sync.Mutex  // 互斥锁

	var wg sync.WaitGroup
	threadCount := 10000
	wg.Add(threadCount)
	for i := 0; i < threadCount; i++ {
		go func() {
			defer wg.Done()
			
			mutex.Lock() // 加锁
			num++ // 临界区
			mutex.Unlock() // 解锁
			
		}()
	}

	wg.Wait()
	fmt.Println(num) // 10000

}

如何使用互斥锁

Mutex 保持 Go 一贯的简洁风格,开箱即用,声明一个变量默认是没有加锁的,加锁使用 Lock() 方法,解锁使用 Unlock() 方法。

使用方式一:直接声明使用

这个在上例中已经体现了,直接看上面的例子就好

使用方式二:封装在其他结构体中

我们可以将 Mutex 封装在 struct 中,封装成线程安全的函数供外部调用。比如我们封装了一个线程安全的计数器,调用 Add() 就加一,调用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
package main

import (
	"fmt"
	"sync"
)


type Counter struct {
	num   int
	mutex sync.Mutex
}

// 加一操作,涉及到临界区 num,加锁解锁
func (counter *Counter) Add() {
	counter.mutex.Lock()
	defer counter.mutex.Unlock()
	counter.num++
}

// 返回数量,涉及到临界区 num,加锁解锁
func (counter *Counter) Count() int {
	counter.mutex.Lock()
	defer counter.mutex.Unlock()
	return counter.num
}

func main() {
	threadCount := 10000
  
	var counter Counter
	var wg sync.WaitGroup
	
	wg.Add(threadCount)
	for i := 0; i < threadCount; i++ {
		go func() {
			defer wg.Done()
			counter.Add()
		}()
	}

	wg.Wait() // 等待所有 goroutine 都执行完
	fmt.Println(counter.Count()) // 10000

}

Go 中,map 结构是不支持并发的,如果并发读写就会 panic

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
// 运行会 panic,提示 fatal error: concurrent map writes
func main() {
	m := make(map[string]string)
	var wait sync.WaitGroup
	wait.Add(1000)

	for i := 0; i < 1000; i++ {
		item := fmt.Sprintf("%d", i)
		go func() {
			wait.Done()
			m[item] = item
		}()
	}
	wait.Wait()
}

基于 Mutex ,我们可以实现一个线程安全的 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
import (
	"fmt"
	"sync"
)

type ConcurrentMap struct {
	mutex sync.Mutex
	items map[string]interface{}
}

func (c *ConcurrentMap) Add(key string, value interface{}) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.items[key] = value
}

func (c *ConcurrentMap) Remove(key string) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	delete(c.items, key)
}
func (c *ConcurrentMap) Get(key string) interface{} {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	return c.items[key]
}

func NewConcurrentMap() ConcurrentMap {
	return ConcurrentMap{
		items: make(map[string]interface{}),
	}
}

func main() {
	m := NewConcurrentMap()
	var wait sync.WaitGroup
	wait.Add(1000)

	for i := 0; i < 1000; i++ {
		item := fmt.Sprintf("%d", i)
		go func() {
			wait.Done()
			m.Add(item, item)
		}()
	}
	wait.Wait()
	fmt.Println(m.Get("100")) // 100
}

当然,基于互斥锁 Mutex 实现的线程安全 map 并不是性能最好的,基于读写锁 sync.RWMutex 和 分片 可以实现性能更好的、线程安全的 map,开发中比较常用的并发安全 maporcaman / concurrent-map

互斥锁的常见问题

从上面可以看出,Mutex 的使用过程方法比较简单,但还是有几点需要注意:

  1. Mutex 是可以在 goroutine A 中加锁,在 goroutine B 中解锁的,但是在实际使用中,尽量保证在同一个 goroutine 中加解锁。比如 goroutine A 申请到了锁,在处理临界区资源的时候,goroutine B 把锁释放了,但是 A 以为自己还持有锁,会继续处理临界区资源,就可能会出现问题。

  2. Mutex 的加锁解锁基本都是成对出现,为了解决忘记解锁,可以使用 defer 语句,在加锁后直接 defer mutex.Unlock();但是如果处理完临界区资源后还有很多耗时操作,为了尽早释放锁,不建议使用 defer,而是在处理完临界区资源后就调用 mutex.Unlock() 尽早释放锁。

 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
// 逻辑复杂,可能会忘记释放锁
func main() {
	var mutex sync.Mutex
	mutex.Lock()

	if *** {
		if *** {
			// 处理临界区资源
			mutex.Unlock()
			return
		}
		// 处理临界区资源
		mutex.Unlock()
		return
	}

	// 处理临界区资源
	mutex.Unlock()
	return
}


// 避免逻辑复杂忘记释放锁,使用 defer语句,成对出现
func main() {
	var mutex sync.Mutex
	mutex.Lock()
	defer mutex.Unlock()

	if *** {
		if *** {
			// 处理临界区资源
			return
		}
		// 处理临界区资源
		return
	}

	// 处理临界区资源
	return
}


  1. Mutex 不能复制使用

Mutex 是有状态的,比如我们对一个 Mutex 加锁后,再进行复制操作,会把当前的加锁状态也给复制过去,基于加锁的 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
26
27
28
package main

import (
	"fmt"
	"sync"
)

type Counter struct {
	mutex sync.Mutex
	num   int
}

func SomeFunc(c Counter) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.num--
}

func main() {
	var counter Counter
	counter.mutex.Lock()
	defer counter.mutex.Unlock()

	counter.num++
	// Go都是值传递,这里复制了 counter,此时 counter.mutex 是加锁状态,在 SomeFunc 无法再次加锁,就会一直等待
	SomeFunc(counter)

}

总结

本篇文章我们简单学习了下 sync.Mutex 的使用方式,以及可能会踩坑的问题,先学着用起来,下篇文章我们一起看源码!

更多

微信公众号:CodePlayer