Go语言sync包的应用详解

在并发编程中同步原语也就是我们通常说的锁的主要作用是保证多个线程或者 goroutine 在访问同一片内存时不会出现混乱的问题。 Go 语言的 sync 包提供了常见的并发编程同步原语，上一期转载的文章《 Golang 并发编程之同步原语 》中也详述了 Mutex 、 RWMutex 、 WaitGroup 、 Once 和 Cond 这些同步原语的实现原理。今天的文章里让我们回到应用层，聚焦 sync 包里这些同步原语的应用场景，同时也会介绍 sync 包中的 Pool 和 Map 的应用场景和使用方法。话不多说，让我们开始吧。

sync.Mutex

sync.Mutex 可能是 sync 包中使用最广泛的原语。它允许在共享资源上互斥访问（不能同时访问）：

mutex := &sync.Mutex{}

mutex.Lock()
// Update共享变量 (比如切片，结构体指针等)
mutex.Unlock()

必须指出的是，在第一次被使用后，不能再对 sync.Mutex 进行复制。（ sync 包的所有原语都一样）。如果结构体具有同步原语字段，则必须通过指针传递它。

sync.RWMutex

sync.RWMutex 是一个读写互斥锁，它提供了我们上面的刚刚看到的 sync.Mutex 的 Lock 和 UnLock 方法（因为这两个结构都实现了 sync.Locker 接口）。但是，它还允许使用 RLock 和 RUnlock 方法进行并发读取：

mutex := &sync.RWMutex{}

mutex.Lock()
// Update 共享变量
mutex.Unlock()

mutex.RLock()
// Read 共享变量
mutex.RUnlock()

sync.RWMutex 允许至少一个读锁或一个写锁存在，而 sync.Mutex 允许一个读锁或一个写锁存在。

通过基准测试来比较这几个方法的性能：

BenchmarkMutexLock-4       83497579         17.7 ns/op
BenchmarkRWMutexLock-4     35286374         44.3 ns/op
BenchmarkRWMutexRLock-4    89403342         15.3 ns/op

可以看到锁定/解锁 sync.RWMutex 读锁的速度比锁定/解锁 sync.Mutex 更快，另一方面，在 sync.RWMutex 上调用 Lock() / Unlock() 是最慢的操作。

因此，只有在频繁读取和不频繁写入的场景里，才应该使用 sync.RWMutex 。

sync.WaitGroup

sync.WaitGroup 也是一个经常会用到的同步原语，它的使用场景是在一个 goroutine 等待一组 goroutine 执行完成。

sync.WaitGroup 拥有一个内部计数器。当计数器等于 0 时，则 Wait() 方法会立即返回。否则它将阻塞执行 Wait() 方法的 goroutine 直到计数器等于 0 时为止。

要增加计数器，我们必须使用 Add(int) 方法。要减少它，我们可以使用 Done() （将计数器减 1 ），也可以传递负数给 Add 方法把计数器减少指定大小， Done() 方法底层就是通过 Add(-1) 实现的。

在以下示例中，我们将启动八个 goroutine ，并等待他们完成：

wg := &sync.WaitGroup{}

for i := 0; i < 8; i++ {
  wg.Add(1)
  go func() {
    // Do something
    wg.Done()
  }()
}

wg.Wait()
// 继续往下执行...

每次创建 goroutine 时，我们都会使用 wg.Add(1) 来增加 wg 的内部计数器。我们也可以在 for 循环之前调用 wg.Add(8) 。

与此同时，每个 goroutine 完成时，都会使用 wg.Done() 减少 wg 的内部计数器。

main goroutine 会在八个 goroutine 都执行 wg.Done() 将计数器变为 0 后才能继续执行。

sync.Map

sync.Map 是一个并发版本的 Go 语言的 map ，我们可以：

• 使用 Store(interface {}，interface {}) 添加元素。
• 使用 Load(interface {}) interface {} 检索元素。
• 使用 Delete(interface {}) 删除元素。
• 使用 LoadOrStore(interface {}，interface {}) (interface {}，bool) 检索或添加之前不存在的元素。如果键之前在 map 中存在，则返回的布尔值为 true 。
• 使用 Range 遍历元素。

m := &sync.Map{}

// 添加元素
m.Store(1, "one")
m.Store(2, "two")

// 获取元素1
value, contains := m.Load(1)
if contains {
  fmt.Printf("%s\n", value.(string))
}

// 返回已存value，否则把指定的键值存储到map中
value, loaded := m.LoadOrStore(3, "three")
if !loaded {
  fmt.Printf("%s\n", value.(string))
}

m.Delete(3)

// 迭代所有元素
m.Range(func(key, value interface{}) bool {
  fmt.Printf("%d: %s\n", key.(int), value.(string))
  return true
})

上面的程序会输出：

one
three
1: one
2: two

如你所见， Range 方法接收一个类型为 func(key，value interface {})bool 的函数参数。如果函数返回了 false ，则停止迭代。有趣的事实是，即使我们在恒定时间后返回 false ，最坏情况下的时间复杂度仍为 O(n) 。

我们应该在什么时候使用 sync.Map 而不是在普通的 map 上使用 sync.Mutex ？

• 当我们对 map 有频繁的读取和不频繁的写入时。
• 当多个 goroutine 读取，写入和覆盖不相交的键时。具体是什么意思呢？例如，如果我们有一个分片实现，其中包含一组4个 goroutine ，每个 goroutine 负责25％的键（每个负责的键不冲突）。在这种情况下， sync.Map 是首选。

sync.Pool

sync.Pool 是一个并发池，负责安全地保存一组对象。它有两个导出方法：

Get() interface{}
Put(interface{})

pool := &sync.Pool{}

pool.Put(NewConnection(1))
pool.Put(NewConnection(2))
pool.Put(NewConnection(3))

connection := pool.Get().(*Connection)
fmt.Printf("%d\n", connection.id)
connection = pool.Get().(*Connection)
fmt.Printf("%d\n", connection.id)
connection = pool.Get().(*Connection)
fmt.Printf("%d\n", connection.id)

输出：

需要注意的是 Get() 方法会从并发池中随机取出对象，无法保证以固定的顺序获取并发池中存储的对象。

还可以为 sync.Pool 指定一个创建者方法：

pool := &sync.Pool{
  New: func() interface{} {
    return NewConnection()
  },
}

connection := pool.Get().(*Connection)

这样每次调用 Get() 时，将返回由在 pool.New 中指定的函数创建的对象（在本例中为指针）。

那么什么时候使用sync.Pool？有两个用例：

第一个是当我们必须重用共享的和长期存在的对象（例如，数据库连接）时。第二个是用于优化内存分配。

让我们考虑一个写入缓冲区并将结果持久保存到文件中的函数示例。使用 sync.Pool ，我们可以通过在不同的函数调用之间重用同一对象来重用为缓冲区分配的空间。

第一步是检索先前分配的缓冲区（如果是第一个调用，则创建一个缓冲区，但这是抽象的）。然后， defer 操作是将缓冲区放回 sync.Pool 中。

func writeFile(pool *sync.Pool, filename string) error {
    buf := pool.Get().(*bytes.Buffer)

  defer pool.Put(buf)

    // Reset 缓存区，不然会连接上次调用时保存在缓存区里的字符串foo
    // 编程foofoo 以此类推
    buf.Reset()

    buf.WriteString("foo")

    return ioutil.WriteFile(filename, buf.Bytes(), 0644)
}

sync.Once

sync.Once 是一个简单而强大的原语，可确保一个函数仅执行一次。在下面的示例中，只有一个 goroutine 会显示输出消息：

once := &sync.Once{}
for i := 0; i < 4; i++ {
    i := i
    go func() {
        once.Do(func() {
            fmt.Printf("first %d\n", i)
        })
    }()
}

我们使用了 Do(func ()) 方法来指定只能被调用一次的部分。

sync.Cond

sync.Cond 可能是 sync 包提供的同步原语中最不常用的一个，它用于发出信号（一对一）或广播信号（一对多）到 goroutine 。让我们考虑一个场景，我们必须向一个 goroutine 指示共享切片的第一个元素已更新。创建 sync.Cond 需要 sync.Locker 对象（ sync.Mutex 或 sync.RWMutex ）：

cond := sync.NewCond(&sync.Mutex{})

然后，让我们编写负责显示切片的第一个元素的函数：

func printFirstElement(s []int, cond *sync.Cond) {
    cond.L.Lock()
    cond.Wait()
    fmt.Printf("%d\n", s[0])
    cond.L.Unlock()
}

我们可以使用 cond.L 访问内部的互斥锁。一旦获得了锁，我们将调用 cond.Wait() ，这会让当前 goroutine 在收到信号前一直处于阻塞状态。

让我们回到 main goroutine 。我们将通过传递共享切片和先前创建的 sync.Cond 来创建 printFirstElement 池。然后我们调用 get() 函数，将结果存储在 s[0] 中并发出信号：

s := make([]int, 1)
for i := 0; i < runtime.NumCPU(); i++ {
    go printFirstElement(s, cond)
}

i := get()
cond.L.Lock()
s[0] = i
cond.Signal()
cond.L.Unlock()

这个信号会解除一个 goroutine 的阻塞状态，解除阻塞的 goroutine 将会显示 s[0] 中存储的值。

但是，有的人可能会争辩说我们的代码破坏了 Go 的最基本原则之一：

不要通过共享内存进行通信；而是通过通信共享内存。

确实，在这个示例中，最好使用 channel 来传递 get() 返回的值。但是我们也提到了 sync.Cond 也可以用于广播信号。我们修改一下上面的示例，把 Signal() 调用改为调用 Broadcast() 。

i := get()
cond.L.Lock()
s[0] = i
cond.Broadcast()
cond.L.Unlock()

在这种情况下，所有goroutine都将被触发。

众所周知， channel 里的元素只会由一个 goroutine 接收到。通过 channel 模拟广播的唯一方法是关闭 channel 。

当一个channel被关闭后，channel中已经发送的数据都被成功接收后，后续的接收操作将不再阻塞，它们会立即返回一个零值。

但是这种方式只能广播一次。因此，尽管存在很大争议，但这无疑是 sync.Cond 的一个有趣的功能。