如果写到并发的程序，就要考虑加锁。而加锁很容易出现 bug，且极难排查。本文以 golang 语言为例，介绍怎样正确地锁住临界区。

一、错误的互斥锁示例

我们以一段测试代码为例，初始化一个 map，我们并发的调用 f 函数，希望用 map 做去重。并且为了减小锁的粒度，我们用了读写锁，读操作互相不阻塞，而写操作互相阻塞，代码如下：

package main

import (
	"github.com/datager/codes/gocodes/dg/utils/signal"
	"sync"
	"time"
)

var (
	mp    = map[string]struct{}{}
	mutex = sync.RWMutex{}
)

func main() {
	go f("goroutine1:", "k")
	go f("goroutine2:", "k")
	signal.WaitForExit()
}

func f(tag string, k string) {
	mutex.RLock()
	_, exist := mp[k]
	mutex.RUnlock()
	if exist {
		println(tag, "already exist")
		return
	}

	// business logic
	time.Sleep(time.Second)

	mutex.Lock()
	mp[k] = struct{}{}
	println(tag, "because not exist, so put into it, now len is", len(mp))
	mutex.Unlock()
}

然而，打印结果如下，我们并发地调用了函数 f，并且用了相同的参数 k，但却并未起到去重的作用，如下结果的第一行 len = 1，而第二行的 len 居然不为 2：

goroutine2: because not exist, so put into it, now len is 1
goroutine1: because not exist, so put into it, now len is 1

这是因为 RLock、RUnlock 和 Lock、Unlock 之间的代码并未受临界区保护：假设以 goroutine1 与 goroutine2 经过 RLock 均得到 !exist 的结果为起点，当 goroutine1 先 Lock 并 UnLock 后，goroutine2 才拿到 Lock 的话，goroutine 2 就会覆盖 goroutine1 的结果，导致 mp[k] = struct{}{} 被错误地执行了2次。

根源是：读是为了写，如果有读操作，且有写操作，则应将读写操作绑定作为一个临界区。而不应该有2个临界区。

二、粗暴的临界区

所以，粗暴地方式，就如上文所说，将读写操作绑定作为一个临界区，代码如下：

package main

import (
	"sync"
	"time"
)

var (
	mp    = map[string]struct{}{}
	mutex = sync.Mutex{}
)

func main() {
	go f("goroutine1:", "k")
	go f("goroutine2:", "k")
	signal.WaitForExit()
}

func f(tag string, k string) {
	mutex.Lock()
	_, exist := mp[k]
	if exist {
		println(tag, "already exist")
		mutex.Unlock()
		return
	}

	// business logic
	time.Sleep(time.Second)

	mp[k] = struct{}{}
	println(tag, "because not exist, so put into it, now len is", len(mp))
	mutex.Unlock()
}

运行后，正确的得到了输出，效果如下：

goroutine1: because not exist, so put into it, now len is 1
goroutine2: already exist

三、double check 来提升性能

然而，为了性能，我们可以用常见的 double check 方式，来处理临界区。

即先无锁检查，再锁 + 判断 + 处理 + 解锁。

通过最先的无锁检查，可以避免没必要的临界区锁定，从而提升性能。代码如下：

package main

import (
	"github.com/datager/codes/gocodes/dg/utils/signal"
	"sync"
	"time"
)

var (
	mp    = map[string]struct{}{}
	mutex = sync.Mutex{}
)

func main() {
	go f("goroutine1:", "k")
	go f("goroutine2:", "k")
	signal.WaitForExit()
}

func f(tag string, k string) {
	_, firstExist := mp[k]
	if firstExist {
		println(tag, "already exist from first check")
		return
	}

	mutex.Lock()
	_, exist := mp[k]
	if exist {
		println(tag, "already exist from second check")
		mutex.Unlock()
		return
	}

	// business logic
	time.Sleep(time.Second)

	mp[k] = struct{}{}
	println(tag, "because not exist, so put into it, now len is", len(mp))
	mutex.Unlock()
}

正确地输出了结果，其中 goroutine1 加锁了，而后续拿到锁的 goroutine2 在无需拿锁的情况下就得到了 firstExist 的结果并提前 return，在结果如下：

goroutine1: because not exist, so put into it, now len is 1
goroutine2: already exist from second check

四、总结

本文总结了锁临界区的常见问题，因为读是为了写，所以需要将读和写放在同一个临界区中来保证正确性，并且为了性能可以用 double check 的方式减少锁的次数。