文章/答案/技术大牛

发布

社区首页 >专栏 >互斥锁与读写锁：如何使用锁完成Go程同步？

互斥锁与读写锁：如何使用锁完成Go程同步？

LIYI

发布于 2021-01-26 08:05:51

1.1K00

代码可运行

文章被收录于专栏：艺述论专栏艺述论专栏

运行总次数：0

代码可运行

图转自https://colobu.com/2018/12/18/dive-into-sync-mutex/

这张图容易让人产生误解，容易让人误以为goroutine1获取的锁，只有goroutine1能释放，其实不是这样的。“秦失其鹿，天下共逐之”。在这张图中，goroutine1与goroutine2竞争的是一种互斥锁。goroutine1成功获取锁以后，锁变成锁定状态，此时goroutine2也可以解锁。

Go语言中有两种锁：

互斥锁 Mutex
读写锁 RWMutex，也叫单写多读锁

第二个锁虽然与第一个仅有两个字母差异，但其实并非同类，稍后我们会看到。名字带有一定的迷惑性，不要被它骗了。

本来Go语言有信道已经足够了，但互斥锁是一种更为常见的多线程协作方式，在其它语言中既然都有实现，Go语言自然也需要支持。

看到锁，我首先想到了一个问题。

Go语言中的锁是怎么实现的？是基于信道实现的吗？

翻一下在官方源码src/sync/mutex.go，Mutex的结构体是这样定义的：

type Mutex struct {
    state int32
    sema  uint32
}

从这种结构体来看，互斥锁并不是基于信道实现的。

实际上，不要看这个结构体很简单，其实锁内部的实现很复杂。

在互斥锁内部有一种自旋操作，所谓自旋对应于CPU的PAUSE指令，就是让CPU空转30个时钟周期，这期间什么也不干，就是为了等着，等着看锁的状态是否能够切换成功。

麻蛋，CPU很闲吗！CPU上班的时候竟然摸鱼，电费也很贵的好吧。

锁是通过一种特殊的对象，让不同线程可以在指定的时间点实现步伐同步；与信道不同的是，信道是不阻塞Go程的，但锁却会。

具体讲，在Go语言中的两种锁中，普通锁Mutex是互斥锁，顾名思义这种锁就像十字路口的红绿灯，一方通行，一方停止，它会直接阻塞Go程；另一种读写锁RWMutex，这种锁是改进的立交桥版本，只阻塞Go程间的写写、读写，但不阻塞读读，稍后会看到这方面具体的实例，体会它们之间的差异。

所以你看，不仅锁不是基于信道实现的，并且性能还比信道差。虽然它在Go语言编程中不被推荐使用，我们还是需要了解一下，这有助于我们有时候阅读别人不太好理解的代码。

普通锁如何使用？

普通锁就是Mutex，它虽然内部复杂，但对外暴露的方法就是两个：

Lock，上锁
Unlock，解锁

什么是上锁，要锁住谁？什么是解锁，又解除对谁的控制？

我们看一段代码吧：

package main 

import ("sync")

var l sync.Mutex
var a string

func f() {
    a = "hi, ly"
    l.Unlock() // Unlock 方法解锁 m，如果 m 未加锁会导致运行时错误。
}

func main() {
    l.Lock() // 默认l是零值解锁状态，在这里先加锁
    go f()
    l.Lock() // l 已经加锁，则阻塞直到 l 解锁。
    println(a) // hi,ly
}

源码见：go-easy/并发/锁/mutex1.go

输出是这样的：

hi，ly

在该示例中，第14行的Lock什么意思，它代表main中开始锁住代码吗？那为什么下面main中没有Unlock的代码？为什么第10行的Unlock的操作却在另一个Go程f()中？

对Go语言中锁的理解，不能像SQL的事务那样，不是”开启事务—>干事—>提交或撤消“这样一个过程：

开启事务 {
    … code
    提交 
} catch(错误) {
    撤消事务
}

互斥锁的Lock、Unlock操作，只针对锁对象本身，并非针对Lock、Unlock之外的那些代码。

互斥锁就是用于同步状态的，或者说是用于同步不同Go程间的事件时间点的。就像十字路口的红绿灯的一样，当灯变成红灯后，下一步如果想让它再变成红灯，必须先把它至少变回一次绿灯；而在此之前，要等待，我们正是利用这种等待的特性，实现了Go程间的同步行为。

具体来讲，在上面示例中，第16行l.Lock发生了阻塞，因为此时l已经处于了Locked状态，除非第10行代码l.Unlock将锁的状态先改变，否则第16行的代码不能继续向下走。

而在这个示例中，并不是说我们在main()中调用了l.Lock（这是一个Go程）、在f()中就不能继续读写内存了（这是另一个Go程），事实上我们在f()中仍然可以对变量a进行自由读写。

使用普通互斥锁，同步的是事件时间点，并没有对“Go程对内存的访问”作任何限制。事实上普通互斥锁也没有这种能力。

有一句教科书式的话是这样说的：对于任何 sync.Mutex 或 sync.RWMutex 类型的变量 l ，满足 a < b ，则我们对 l.Unlock() 的第 a 次调用，总是在对 l.Lock() 的第 b 次调用返回前发生。

简单理解这句话，就两条规则：

Unlock要发生在Lock之前
如果尚未Lock，直接Unlock，则会抛出异常

有了这句教科书真言，算是如获至宝了，本身它也适用于RWMutex。如果我们想使用RWMutex改写上面的示例，应当如何改写呢？

看一下代码吧：

package main 

import ("sync")

var l sync.RWMutex
var a string = "hi"

func f() {
    // println(a)
    a = "hi, ly"
    l.Unlock() 
}

func main() {
    l.Lock() 
    go f()
    l.Lock() 
    println(a) // hi,ly
}

源码见：go-easy/并发/锁/mutex1-1.go

输出是一样的。

我们仅是在第5行改变了一下变量l的类型，RWMutex也可以当作普通的Mutex使用。

那么加强版本的RWMutex还有哪些其它妙用呢？

如何使用加强版本的读写锁？

普通锁并不能满足所有场景的互斥需求。看一张表格：

	读	写
读	读读 √	读写 x
写	写读 x	写写 x

有时候我们有多个线程，譬如简单一些有两个线程，我们要限制它们同时写，但不限制它们同时读。这也很容易理解，这种场景多发生在数据库操作或文件操作中。大多数情况下，读表比写表要快，因为读表是可以并发的，而写表因为要力保数据一致，是要锁表的，会产生阻塞。

接下来我们看看一下读写锁的示例吧：

package main

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

func main() {
    var l sync.RWMutex
    var data = 1

    for i := 0; i < 10; i++ {
        go func(t int) {
            l.RLock()
            defer l.RUnlock()
            fmt.Printf("Read data: %d %v\n", t, data)
        }(i)
        // if i == 3 {
        //     time.Sleep(time.Second)
        // }
        go func(t int) {
            l.Lock()
            defer l.Unlock()
            data++
            fmt.Printf("Write Data: %d %v \n", t, data)
        }(i)
    }
    time.Sleep(time.Second)
}

源码见：go-easy/并发/锁/mutex2.go

输出：

Read data: 0 1
Write Data: 1 2 
Read data: 1 2
Read data: 5 2
Read data: 7 2
Read data: 8 2
Read data: 9 2
Read data: 2 2
Read data: 3 2
Read data: 4 2
Read data: 6 2
Write Data: 0 3 
Write Data: 3 4 
Write Data: 4 5 
Write Data: 5 6 
Write Data: 7 7 
Write Data: 6 8 
Write Data: 8 9 
Write Data: 9 10 
Write Data: 2 11

需要指出的是，这个输出并不是固定的。第一行第一次Read data输出的data有可能是1，也有很大概率是2。为什么输出不固定？当环境一致、输入条件一致时，电脑输出不应该固定吗？电脑不是最诚实的吗？

单线程时电脑确实很诚实，多线程时就不一定了。电脑是人设计的，这方面可能也承袭了人类的缺陷。人类一男一女谈恋爱比较甜蜜简单，多女同追一男，或多男同追一女就容易发生口角或战争。

回到上面的问题，其实不是的，因为本质上这些Go程它们是并发的。第25行data自增代码的执行时间点会与谁对齐，并不固定，完全看当时CPU的心情。

但有一些规范仍然是固定的，譬如：对于任何 sync.RWMutex 类型的变量 l 对 l.RLock 的调用，存在一个这样的 n**，使得 l.RLock 在对 l.Unlock 的第 n 次调用之后发生（返回），且与其相匹配的 l.RUnlock 在对 l.Lock的第 n+1 次调用之前发生。**

这句教科书的话理解起来特别费劲，画个图表就是这样的：

Lock … Unlock ….. RLock ... RUnlock … [RLock …… RUnlock …] Lock … Unlock ...

在读写锁上，先明确一下，Lock与Unlock是写的上锁与解锁，RLock与RUnlock是读的上锁与解锁。它只有这4个方法，它没有WLock与WUnlock。

读写锁在读上是不互斥的。所以它允许多个Go程同时RLock与RUnlock，这是合法的；但是一但有一个线程进行了Lock上写锁，所有的读都要停下来，此时Lock就是一个同步的时间点，走过Unlock后，RLock与RUnlock又可以开始活跃了。

读写锁的这种机制有点像中国古代的婚姻制度三妻四妾，家里妻妾成群好比读锁并飞好不热闹，男人好比写锁，男人一来所有妻妾就闭嘴了。

如果我们把mutex2.go中的第19~21行的代码反注释一下，大体输出就会变成这样了：

Read data: 0 1
Read data: 2 1
Write Data: 1 2 
Read data: 3 2
Read data: 1 2
Write Data: 2 3 
Write Data: 0 4 
Read data: 5 4
Read data: 4 4
Write Data: 4 5 
Read data: 6 5
Read data: 7 5
Read data: 8 5
Read data: 9 5
Write Data: 3 6 
Write Data: 9 7 
Write Data: 6 8 
Write Data: 7 9 
Write Data: 5 10 
Write Data: 8 11

我们看到在这个输出里面，因为在第20行人为添加了休眠时间，将某些读线程与写线程隔开了。但从打印行为上来看，写线程成为了读线程的分隔点。在写线程改变data变量以后，读线程总是能读到改变之后的值。这和数据库的读取写入是同样的道理，改变效果总能得到及时彰显。

在这里有个问题我们思考一下，在第14行开启的读线程内，不可以向内存写入数据吗？

并不是的。我们看一个稍加改造之后的示例代码：

package main

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

func main() {
    var l sync.RWMutex
    var data = 1.0

    for i := 0; i < 10; i++ {
        go func(t int) {
            l.RLock()
            defer l.RUnlock()
            data += .1
            fmt.Printf("Read data: %d %v\n", t, data)
        }(i)
        // if i == 3 {
        //     time.Sleep(time.Second)
        // }
        go func(t int) {
            l.Lock()
            defer l.Unlock()
            data++
            fmt.Printf("Write Data: %d %v \n", t, data)
        }(i)
    }
    time.Sleep(time.Second)
}

源码见：go-easy/并发/锁/mutex2-1.go

第11行将data的默认值修改为1.0，此时它不再是整形了。还有，添加了第17行代码，现在读线程也开始尝试向内存里写入数据了。输出结果是这样的：

Read data: 0 1.1
Write Data: 1 2.1 
Read data: 2 2.2
Read data: 7 2.4000000000000004
Read data: 8 2.5000000000000004
Read data: 6 2.3000000000000003
Read data: 4 2.8000000000000007
Read data: 3 2.900000000000001
Read data: 1 3.000000000000001
Read data: 5 2.7000000000000006
Read data: 9 2.6000000000000005
Write Data: 9 4.000000000000001 
Write Data: 2 5.000000000000001 
Write Data: 3 6.000000000000001 
Write Data: 4 7.000000000000001 
Write Data: 5 8 
Write Data: 6 9 
Write Data: 7 10 
Write Data: 8 11 
Write Data: 0 12

我们看到，即使是“读”线程，也能写入数据。如果说示例mutex2.go演示的是“多读一写”场景，这个mutex2-1.go示例实际演示的却是“多写”场景。

所以我们看到，虽然“读”线程打印的data并不是严格按照从小到大的顺序打印的，譬如第5行2.5比第6行2.3还要大，因为本质上它们是并发执行的，结果是随机的。但data却是以0.1的步伐均匀递增的，看第2~11行，data从2.2按照0.1的步伐均匀递增到3.0。那一长串零最后面的数字是由于计算精度造成的，可以忽略。

这是为什么？

因为在第17行我们写内存了。第17行代码所在的Go程虽然开启的是读锁，但实际上代码进行了写入，此时的并发场景不是“读读”，而是“写写”了。我们只需要将第17行的代码注释掉，再看一看它的表现就明白了：

package main

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

func main() {
    var l sync.RWMutex
    var data = 1.0

    for i := 0; i < 10; i++ {
        go func(t int) {
            l.RLock()
            defer l.RUnlock()
            // data += .1
            fmt.Printf("Read data: %d %v\n", t, data)
        }(i)
        // if i == 3 {
        //     time.Sleep(time.Second)
        // }
        go func(t int) {
            l.Lock()
            defer l.Unlock()
            data++
            fmt.Printf("Write Data: %d %v \n", t, data)
        }(i)
    }
    time.Sleep(time.Second)
}

源码见：go-easy/并发/锁/mutex2-2.go

输出：

Read data: 0 1
Write Data: 2 2 
Read data: 1 2
Read data: 6 2
Read data: 9 2
Read data: 7 2
Read data: 3 2
Read data: 2 2
Read data: 5 2
Read data: 4 2
Read data: 8 2
Write Data: 1 3 
Write Data: 0 4 
Write Data: 3 5 
Write Data: 4 6 
Write Data: 6 7 
Write Data: 5 8 
Write Data: 7 9 
Write Data: 8 10 
Write Data: 9 11

看看现在的输出，读锁完全并发了，它们挤在一块执行，只拿到了data等于2。

所以我们看，在使用读写锁时，如果我们向内存写入了，此时开启RLock、与开启Lock是一样的。不了解这一点机制，很容易就写出错误的代码，当然了别人的代码也不易读懂。

互斥锁的早期源码

虽然最新的Mutex源码很复杂，难于理解，但早期的Mutex源码却很简单，可谓是骨骼清奇：

package main

type Mutex struct {
    key  int32
    sema int32
}

func xadd(val *int32, delta int32) (new int32) {
    for {
        v := *val
        // cas这个函数是原子操作，它有三个参数，第一个是目标数据的地址，第二个是目标数据的旧值，第三个则是等待更新的新值。每次CAS都会用old和addr内的数据进行比较，如果数值相等，则执行操作，用new覆盖addr内的旧值，如果数据不相等，则忽略后面的操作。
        if cas(val, v, v+delta) {
            return v + delta
        }
    }
    panic("unreached")
}

func (m *Mutex) Lock() {
    if xadd(&m.key, 1) == 1 {
        // changed from 0 to 1; we hold lock
        return
    }
    // semacquire函数首先检查信号量是否为0：如果大于0，让信号量减一，返回；
    // 如果等于0，就调用goparkunlock函数，把当前Goroutine放入该sema的等待队列，并把他设为等待状态。
    sys.semacquire(&m.sema)
}

func (m *Mutex) Unlock() {
    if xadd(&m.key, -1) == 0 {
        // changed from 1 to 0; no contention
        return
    }
    // semrelease函数首先让信号量加一，然后检查是否有正在等待的Goroutine：如果没有，直接返回；
    // 如果有，调用goready函数唤醒一个Goroutine。
    sys.semrelease(&m.sema)
}

源码见：go-easy/并发/锁/SimpleMutex.go

这份源码已经加了注释，很好理解。后来变得复杂，是为了解决多并发线程中容易出现的尾部延迟现象，加入了饥饿模式。有了这种机制，更加加强了并发微线程执行的不确定性。不一定后来的微线程就启动的晚，也不一定早期的微线程就一直没有机会。我们所以看到，mutex2-2.go示例中第18行代码的执行像完全随机一样。理解这种机制就好。

一道题：看懂这道题就理解基本的互斥锁了

我们看一道有意思的题：

package main
import (
    "sync"
    "time"
)
func main() { // g1
    var mu sync.Mutex
    go func() { // g2
        mu.Lock()
        time.Sleep(10 * time.Second)
        mu.Unlock()
    }()
    time.Sleep(time.Second)
    mu.Unlock()
    select {}
}