并发编程并发中互斥锁常见问题总结

Posted 2021-11-21 了 凡

博主介绍：

– 我是了 凡，喜欢每日在简书上投稿日更的读书感悟笔名：三月_刘超。专注于 Go Web 后端，了解过一些Python、Java、算法、前端等领域。微信公众号【了凡银河系】期待你的关注，企鹅群号（798829931）。未来大家一起加油啊~

---

前言

这里介绍常见四种错误：
一：Lock/Unlock不是成队出现
二：Copy已使用的Mutex
三：重入
四：死锁

---

文章目录

• 前言
• Lock/Unlock不是成对出现
• Copy 已使用的 Mutex
• • 长期解决方法
  • vet 工具的实现原理
• 重入
• • 可重入锁
  • Mutex使用的错误场景
  • Mutex可重入锁实现
  • • 方案一：goroutine id
    • 方案二：token
• 死锁
• 总结

---

Lock/Unlock不是成对出现

Lock/Unlock如果没有成队出现就会出现死锁状态，或者因为Unlock一个未加锁的Mutex而导致panic。

缺少Unlock的时候，常见有三种情况：
1.代码中有太多的 if-else 分支，可能在某个分支中漏写了Unlock；
2.在重构的时候把Unlock给删除了；
3.Unlock误写成了Lock。

以上者三种情况，锁被获取后，没能释放就意味着，其他的goroutine（协程）永远无法获取到锁。

缺少Lock的时候

缺少Lock的时候，没有经进行Lock而是直接Unlock这样的情况就会造成一个为加锁的Mutex会panic：

func foo(){
    var mu sync.Mutex
    defer mu.Unlock()
    fmt.Println("hello world!")
}

运行结果：

---

Copy 已使用的 Mutex

Copy 已使用的 Mutex错误，首先就是强调一点同步原语使用后是不可以进行Copy进行复制的。例如Package sync的同步原语使用后就不能进行复制操作，而Mutex是最常用一个同步原语，就不可以复制。但是为什么呢？

原因是 Mutex是一个有状态的对象，它的state字段记录这个锁的状态。如果复制一个已经加锁的Mutex给一个新的变量，那么新的变量，那么新的刚初始化的变量居然被加锁了，者显然不符合期望的，我们肯定期望是一个零值的Mutex。关键是在并发环境下，你根本不知道要复制的Mutex状态是什么，因为要复制的Mutex是由其它goroutine并发访问的，状态可能总是在变换。

例如，举个例子。

type Counter struct {
    sync.Mutex
    Count int
}
func main(){
    var c Counter
    c.Lock()
    defer c.Unlock()
    c.Count++
    foo(c) // 复制锁
}

// 这里Counter的参数是通过复制的方式传入的
func foo(c Counter){
    c.Lock()
    defer c.Unlock()
    fmt.Println("in foo")
}

运行结果：

第十二行在调用foo函数的时候，调用者会复制Mutex变量c作为foo函数的参数，不幸的是，复制之前已经使用了这个锁，这样就导致复制的Counter是一个带状态Counter。

长期解决方法

为了能够及时发现问题，可以使用 vet工具，把检查写在Makefile 文件中，在持续集成的时候跑一跑，这样就可以及时发现问题，及时修复。可以使用 go vet检查这个Go文件。

go vet xxx.go

例如：

可以看到显示的还是很清楚的，是在调用foo函数的时候发生了lock value 复制的情况，还指出问题的代码行数以及copy lock导致的错误。

vet 工具的实现原理

通过 copylock 分析器静态分析实现的。这个分析器会分析函数调用、range遍历、复制、声明、函数返回值等位置，有没有锁的值copy的情景，以此来判断有没有问题。可以说，只要实现了Locker 接口，就会被分析。具体什么类型会被分析，其实就是实现了Lock/Unlock 两个方法的Locker接口：

var lockerType *types.Interface

// Construct a sync.Locker interface type.
func init(){
    nullary := types.NewSignature(nil, nil, nil, false) // func()
    methods := []*types.Func{
        types.NewFunc(token.NoPos, nil, "Lock", nullary),
        types.NewFunc(token.NoPos, nil, "Unlock", nullary),
    }
    lockerType = types.NewInterface(methods, nil).Complete()
}

当然，有些没有实现Locker接口的同步原语(比如 WaitGroup)，也能分析。具体如何实现目前我还不知道。

---

重入

可重入锁

在 Java 中有一个ReentrantLock，就是可重入锁，这是Java并发包中非常常用的一个同步原语。它的基本行为和互斥锁相同，但是加了一些扩展功能。如果不了解Java也没关系，只是稍微提一下。

具体可重入锁怎么回事呢？就是当一个线程获取锁时，如果没有其他线程拥有这个锁，那么，这个线程就成功获取到这个锁。之后，如果其它线程再请求这个锁，就会处于阻塞等待状态。但是，如果拥有这把锁的线程再请求这把锁的话，不会阻塞，而是成功返回，所以叫可重入锁（有时候也可以叫做递归锁）。只要你拥有这把锁，你可以尽管调用，比如通过递归实现一些算法，调用者不会阻塞或者死锁。

Mutex使用的错误场景

有一个重点，Mutex使用的错误场景，Mutex不是可重入锁。因为Mutex的实现中没有记录哪个goroutine拥有这把锁。理论上，任何 goroutine 都可以随意地Unlock这把锁，所以没办法计算重入条件。

误用Mutex的重入，就会导致报错。例如：

func foo(l sync.Locker) {
   fmt.Println("in foo")
   l.Lock()
   bar(l)
   l.Unlock()
}

func bar(l sync.Locker)  {
   l.Lock()
   fmt.Println("in bar")
   l.Unlock()
}

func main()  {
   l := &sync.Mutex{}
   foo(l)
}

执行这个Mutex重入的例子后，就会发现类似下面的错误。程序一直在请求锁，但是一直没办法获取到锁，结果就是Go运行时发现的死锁了，没有其它地方能够释放锁让程序运行下去，你通过下面的错误堆栈信息就能定位到哪一行阻塞请求锁。

运行结果：

这样可以知道标准库Mutex不是可重入锁，我们想一下自己实现一个。

Mutex可重入锁实现

在实现的过程中，一个关键的问题就是，实现的锁要能记住当前哪个goroutine持有这个锁。两个方案

方案一：通过 hacker 的方式获取到goroutine id，记录下获取锁的goroutine id，它可以实现Locker接口。

方案二：调用Lock/Unlock方法时，由goroutine提供一个token，用来标识自己，而不是我们通过hacker 的方式获取到goroutine id，但是，这样一来，就不满足Loker接口了。

可重入锁（递归锁）解决了代码重入或者递归调用带来的死锁问题，同时它也带来了另一个好处，就是我们可以要求，只有持有锁的 goroutine 才能 Unlock 这个锁。还是毕竟容易实现的，因为在上面这两个方案中，都已经记录了是哪一个goroutine持有这个锁。

方案一：goroutine id

关键第一步是 runtime.Stack方法获取栈帧信息，栈帧信息里包含goroutine id。你可以看看上面panic时候的贴图，goroutine id明明白白的现实着。runtime.Stack 方法可以获取当前的goroutine信息，第二个参数true会输出所有的goroutine信息，信息的格式为：

goroutine 1 [running]:
main.main()
    ...../main.go:19 + 0xb1

第一行格式为goroutine XXX, 其中XXX就是goroutine id，你只要解析出这个id即可。解析的方法可以采用下面的代码：

func GoID() int {
    var buf [64]byte
    n := runtime.Stack(buf[:], false)
    // 得到id字符串
    idField := strings.Fields(strings.TrimPrefix(string(buf[:n]),"goroutine"))
    id, err := strconv.Atoi(idField)
    if err != nil {
        panic(fmt.Sprintf("cannot get goroutine id: %v", err))
    }
    return id
}

看一下hacker的方式，着也是方案一采取的方式。

首先，我们获取运行时的g指针，反解出对应的g的结构。每个运行的goroutine结构的g指针保存在当前goroutine的一个叫做TLS对象中。

第一步：先获取到TLS对象；
第二步：再从TLS中获取goroutine结构的go指针；
第三步：再从g指针中取出goroutine id。

需要注意的是，不同Go版本的goroutine的结构可能不同，所以需要根据Go的不同版本进行调整。想要搞清楚各个版本的goroutine结构体差异，所要涉及的内容又过于底层而且复杂，学习成本太高。所以我们可以重点关注一些库。直接使用第三方的库来获取goroutine id就可以了。

现在已经有很多成熟的方法了，可以支持多个Go版本的goroutine id，有一个常用的库：petermattis/goid

上面描述了如何获取goroutine id，就剩下最后关键一步了，实现一个可以使用的可重入锁：

// RecursiveMutex 包装一个Mutex，实现可重入
type RecursiveMutex struct {
    sync.Mutex
    owner int64 // 当前持有锁的goroutine id
    recursion int32 // 这个goroutine 重入的次数
}

func (m *RecursiveMutex) Lock(){
    gid := goid.Get()
    // 如果当前持有锁的goroutine就是这次调用的goroutine，说明是重入
    if atomic.LoadInt64(&m.owner) == gid {
        m.recursion++
        return
    }
    m.Mutex.Lock()
    // 获得锁的goroutine第一次调用，记录下它的goroutine id，调用次数加1
    atomic.StoreInt64(&m.owner, gid)
    m.recursion = 1
}

func (m *RecursiveMutex) Unlock() {
    gid := goid.Get()
    // 非持有锁的goroutine尝试释放锁，错误的使用
    if atomic.LoadInt64(&m.owner) != gid {
        panic(fmt.Sprintf("wrong the owner(%d): %d!", m.owner, gid))
    }
    // 调用次数减1
    m.recursion --
    if m.recursion != 0 { // 如果这个goroutine还没有完全释放，则直接返回
        return
    }
    // 此goroutine最后一次调用，需要释放锁
    atomic.StoreInt64(&m.owner, -1)
    m.Mutex.Unlock()
}

这个代码是可以直接用的。这个实现也非常巧妙，它相当于给Mutex 打一个补丁，解决了记录锁的持有者的问题。可以看到，我们用owner字段，记录当前锁的拥有者 goroutine 的id；recursion 是辅助字段，用于记录重入的次数。

但是有一点，尽管拥有者可以多次调用Lock，但是也必须调用相同次数的Unlock，这样才能把锁释放掉。这是一个合理的设计，可以保证Lock和Unlock 一一对应。

方案二：token

上一个方案用goroutine id做goroutine 的标识，我们也可以让goroutine 自己来提供标识。当然了，Go开发者不期望你利用goroutine id做一些不确定的东西，所以，他们没有暴露获取goroutine id的方法。

第二种方案

第一步：调用者自己提供一个token，获取锁的时候把这个token传入。
第二步：释放的时候也需要把这个token传入。
第三步：通过用户传入的token替换方案一中goroutine id，其它逻辑和方案一一致。

// Token方式的递归锁
type TokenRecursiveMutex struct{
    sync.Mutex
    token int64
    recursion int32
}

// 请求锁，需要传入token
func (m *TokenRecursiveMutex) Lock(token int64) {
    if atomic.LoadInt64(&m.token) == token{ // 如果传入的token和持有锁的token一致，说明是递归调用
        m.recursion++
        return
    }
    m.Mutex.Lock() // 传入的token不一致，说明不是递归调用
    // 抢到锁之后记录这个token
    atomic.StoreInt64(&m.token, token)
    m.recursion = 1
}

// 释放锁
func (m *TokenRecursiveMutex) Unlock(token int64){
    if atomic.LoadInt64(&m.token) != token { // 释放其它token持有的锁
        panic(fmt.Sprintf("wrong the owner(%d): %d", m.token, token))
    }
    m.recursion-- // 当前持有这个锁的token释放锁
    if m.recursion != 0 { // 还没有回退到最初的递归调用
        return
    }
    atomic.StoreInt64(&m.token, 0) // 没有递归调用了，释放锁
    m.Mutex.Unlock()
}

---

死锁

什么是死锁？两个或两个以上的进程（或线程，goroutine）在执行过程中，因争夺共享资源而处于一种互相等待的状态，如果没有外部干涉，它们都将无法推进下去，此时，我们称系统处于死锁状态或系统产生了死锁。

死锁产生的必要条件。如何避免死锁？只要记住以下一个破坏这四个条件中的一个或者几个，就可以。

• 互斥：至少一个资源是被排他性独享的，其他线程必须处于等待状态，直到资源被释放。
• 持有和等待：goroutine 持有一个资源，并且还在请求其它goroutine 持有的资源，也就是咱们常说的“吃着碗里，看着锅里”的意思。
• 不可剥夺：资源只能由持有它的goroutine来释放。
• 环路等待：一般来说，存在一组等待进程，P={P1,P2,…,PN}，P1等待P2持有的资源，P2等待P3持有的资源，依次类推，最后是PN等待P1持有的资源，这就形成了一个环路等待的死结。

推荐一个操作系统里面在第二章讲解进程的时候有一个哲学家就餐问题，很有意思，可以了解一下。死锁的问题生活中很常见，用心观察。

例如：

去派出所开证明，派出所要求物业先证明自己是本物业的业主，但是，物业要我提供派出所的证明，才能给我开物业证明，结果就陷入了死锁状态。你可以把派出所和物业看成两个goroutine，派出所证明和物业证明是两个资源，双方都持有自己的资源而要求对方的资源，而且自己的资源自己持有，不可剥夺。

最简单的只有两个goroutine相互等待的死锁的例子：

func main() {
    // 派出所证明
    var psCertificate sync.Mutex
    // 物业证明
    var propertyCertificate sync.Mutex
    
    var wg sync.WaitGroup
    wg.Add(2) // 需要派出所和物业都处理
    
    // 派出所处理goroutine
    go func(){
        defer wg.Done() // 派出所处理完成
        
        psCertificate.Lock()
        defer psCertificate.Unlock()
        
        // 检查材料
        time.Sleep(5*time.Second)
        // 请求物业的证明
        propertyCertificate.Lock()
        propertyCertificate.Unlock()
    }()
    
    // 物业处理goroutine
    go func() {
        defer wg.Done() // 物业处理完成
        
        propertyCertificate.Lock()
        defer propertyCertificate.Unlock()
        
        // 检查材料
        time.Sleep(5 * time.Second)
        // 请求派出所的证明
        psCertificate.Lock()
        psCertificate.Unlock()
    }()
    
    wg.Wait()
    fmt.Println("成功完成")
}

这个程序没有办法运行成功，因为派出所的处理和物业的处理是一个环路等待的死结。

Go运行时，有死锁探测的功能，能够检查出是否出现了死锁的情况，如果出现了，就需要调整策略来处理了。

可以引入一个第三方的锁，都依赖这个锁进行业务处理，比如现在政府推行的一站式政务服务中心。或者是解决持有等待问题，物业不需要看到派出所的证明才给开物业证明，等等。

总结

学习了Mutex的一些易错场景，而且，分析了流行的Go开源项目的错误，也分享了开发中的经验总结。需要强调的是，手误和重入导致的死锁，是最常见的使用Mutex的 Bug。

Go 死锁探测工具只能探测整个程序是否因为死锁而冻结了，不能检测出一组goroutine死锁导致的某一块业务冻结的情况。你还可以通过Go运行时自带的死锁检测工具，或者是第三方的工具(比如@go-deadlock、go-tools)进行检查，这样可以尽早发现一些死锁的问题。不过，有些时候，死锁在某些特定情况下才会被触发，所以，如果你的测试或者短时间的运行没问题，不代表程序一定不会有死锁问题。

并发程序最难跟踪调试的就是很难重现，因为并发问题不是按照我们指定的顺序执行的，由于计算机调度的问题和事件触发的时机不同，死锁的Bug可能会在极端的情况下出现。通过搜索日志、查看日志，我们能够知道程序有异常了，比如某个流程一直没有结束。这个时候，可以通过Go pprof工具分析，它提供了一个block profiler监控阻塞的goroutine。除此之外，我们还可以查看全部的goroutine的堆栈信息，通过它，你可以查看阻塞的groutine究竟阻塞在哪一行哪一个对象上了。

---

这次就先讲到这里，如果想要了解更多的golang语言内容一键三连后序每周持续更新！

---