[[409469]]

本文轉載自微信公眾號「Golang技術分享」，作者機器鈴砍菜刀。轉載本文請聯系Golang技術分享公眾號。

Go 語言通過 go 關鍵字開啟 goroutine 讓開發者可以輕松地實現并發編程，而并發程序的有效運行，往往離不開 sync 包的保駕護航。目前，sync 包的賦能列表包括：sync.atomic 下的原子操作、sync.Map 并發安全 map、sync.Mutex 與 sync.RWMutex 提供的互斥鎖與讀寫鎖、sync.Pool 復用對象池、sync.Once 單例模式、 sync.Waitgroup 的多任務協作模式、sync.Cond 的監視器模式。當然，除了 sync 包，還有封裝層面更高的 channel 與 context。

要想寫出合格的 Go 程序，以上的這些并發原語是必須要掌握的。對于大多數 Gopher 而言，sync.Cond 應該是最為陌生，本文將一探究竟。

初識 sync.Cond

sync.Cond 字面意義就是同步條件變量，它實現的是一種監視器(Monitor)模式。

In concurrent programming(also known as parallel programming), a monitor is a synchronization construct that allows threads to have both mutual exclusion and the ability to wait (block) for a certain condition to become false.

對于 Cond 而言，它實現一個條件變量，是 goroutine 間等待和通知的點。條件變量與共享的數據隔離，它可以同時阻塞多個 goroutine，直到另外的 goroutine 更改了條件變量，并通知喚醒阻塞著的一個或多個 goroutine。

初次接觸的讀者，可能會不太明白，那么下面我們看一下 GopherCon 2018 上《Rethinking Classical Concurrency Patterns》 中的演示代碼例子。

 1type Item = int 
 2 
 3type Queue struct { 
 4   items     []Item 
 5   itemAdded sync.Cond 
 6} 
 7 
 8func NewQueue() *Queue { 
 9   q := new(Queue) 
10   q.itemAdded.L = &sync.Mutex{} // 為 Cond 綁定鎖 
11   return q 
12} 
13 
14func (q *Queue) Put(item Item) { 
15   q.itemAdded.L.Lock() 
16   defer q.itemAdded.L.Unlock() 
17   q.items = append(q.items, item) 
18   q.itemAdded.Signal()        // 當 Queue 中加入數據成功，調用 Singal 發送通知 
19} 
20 
21func (q *Queue) GetMany(n int) []Item { 
22   q.itemAdded.L.Lock() 
23   defer q.itemAdded.L.Unlock() 
24   for len(q.items) < n {     // 等待 Queue 中有 n 個數據 
25      q.itemAdded.Wait()      // 阻塞等待 Singal 發送通知 
26   } 
27   items := q.items[:n:n] 
28   q.items = q.items[n:] 
29   return items 
30} 
31 
32func main() { 
33   q := NewQueue() 
34 
35   var wg sync.WaitGroup 
36   for n := 10; n > 0; n-- { 
37      wg.Add(1) 
38      go func(n int) { 
39         items := q.GetMany(n) 
40         fmt.Printf("%2d: %2d\n", n, items) 
41         wg.Done() 
42      }(n) 
43   } 
44 
45   for i := 0; i < 100; i++ { 
46      q.Put(i) 
47   } 
48 
49   wg.Wait() 
50} 

在這個例子中，Queue 是存儲數據 Item 的結構體，它通過 Cond 類型的 itemAdded 來控制數據的輸入與輸出。可以注意到，這里通過 10 個 goroutine 來消費數據，但它們所需的數據量并不相等，我們可以稱之為 batch，依次在 1-10 之間。之后，逐步添加 100 個數據至 Queue 中。最后，我們能夠看到 10 個 gotoutine 都能被喚醒，得到它想要的數據。

程序運行結果如下

1 6: [ 7  8  9 10 11 12] 
2 5: [50 51 52 53 54] 
3 9: [14 15 16 17 18 19 20 21 22] 
4 1: [13] 
5 2: [33 34] 
6 4: [35 36 37 38] 
7 3: [39 40 41] 
8 7: [ 0  1  2  3  4  5  6] 
9 8: [42 43 44 45 46 47 48 49] 
010: [23 24 25 26 27 28 29 30 31 32] 

當然，程序每次運行結果都不會相同，以上輸出只是某一種情況。

sync.Cond 實現

在 $GOPATH/src/sync/cond.go 中，Cond 的結構體定義如下

1type Cond struct { 
2   noCopy noCopy 
3   L Locker 
4   notify  notifyList 
5   checker copyChecker 
6} 

其中，noCopy 與 checker 字段均是為了避免 Cond 在使用過程中被復制，詳見小菜刀的 《no copy 機制》 一文。

L 是 Locker 接口，一般該字段的實際對象是 *RWmutex 或者 *Mutex。

1type Locker interface { 
2   Lock() 
3   Unlock() 
4} 

notifyList 記錄的是一個基于票號的通知列表，這里初次看注釋看不懂沒關系，和下文來回連貫著看。

1type notifyList struct { 
2   wait   uint32         // 用于記錄下一個等待者 waiter 的票號 
3   notify uint32         // 用于記錄下一個應該被通知的 waiter 的票號 
4   lock   uintptr        // 內部鎖 
5   head   unsafe.Pointer // 指向等待者 waiter 的隊列隊頭 
6   tail   unsafe.Pointer // 指向等待者 waiter 的隊列隊尾 
7} 

其中，head 與 tail 是指向 sudog 結構體的指針，sudog 是代表的處于等待列表的 goroutine，它本身就是雙向鏈表。值得一提的是，在 sudog 中有一個字段 ticket 就是用于給當前 goroutine 記錄票號使用的。

Cond 實現的核心模式為票務系統(ticket system)，每一個想要來買票的 goroutine (調用Cond.Wait())我們稱之為 waiter，票務系統會給每個 waiter 分配一個取票碼，等供票方有該取票碼的號時，就會喚醒 waiter。賣票的 goroutine 有兩種，第一種是調用 Cond.Signal() 的，它會按照票號喚醒一個買票的 waiter (如果有的話)，第二種是調用 Cond.Broadcast() 的，它會通知喚醒所有的阻塞 waiter。為了方便讀者能夠比較輕松地理解票務系統，下面我們給出圖解示例。

在 上文中，我們知道 Cond 字段中 notifyList 結構體是一個記錄票號的通知列表。這里將 notifyList 比作排隊取票買電影票，當 G1 通過 Wait 來買票時，發現此時并沒有票可買，因此他只能阻塞等待有票之后的通知，此時他手上已經取得了專屬取票碼 0。同樣的，G2 和 G3 也同樣無票可買，它們分別取到了自己的取票碼 1和 2。而 G4 是電影票提供商，它是賣票的，它通過兩次 Signal 先后帶來了兩張票，按照票號順序依次通知了 G1 和 G2 來取票，并把 notify 更新為了最新的 1。G5 也是買票的，它發現此時已經無票可買了，拿了自己的取票碼 3 ，就阻塞等待了。G6 是個大票商，它通過 Broadcast 可以滿足所有正在等待的買票者都買到票，此時等待的是 G3 和 G5，因此他直接喚醒了 G3 和 G5，并將 notify 更新到和 wait 值相等。

理解了上述取票系統的運作原理后，我們下面來看 Cond 包下四個實際對外方法函數的實現。

NewCond 方法

1func NewCond(l Locker) *Cond { 
2   return &Cond{L: l} 
3} 

用于初始化 Cond 對象，就是初始化控制鎖。

Cond.Wait 方法

1func (c *Cond) Wait() { 
2   c.checker.check() 
3   t := runtime_notifyListAdd(&c.notify) 
4   c.L.Unlock() 
5   runtime_notifyListWait(&c.notify, t) 
6   c.L.Lock() 
7} 

runtime_notifyListAdd 的實現在 runtime/sema.go 的 notifyListAdd ，它用于原子性地增加等待者的 waiter 票號，并返回當前 goroutine 應該取的票號值 t 。runtime_notifyListWait 的實現在runtime/sema.go 的 notifyListWait，它會嘗試去比較此時 goroutine 的應取票號 t 與 notify 中記錄的當前應該被通知的票號。如果 t 小于當前票號，那么直接能得到返回，否則將會則塞等待，通知取號。

同時，這里需要注意的是，由于在進入 runtime_notifyListWait 時，當前 goroutine 通過 c.L.Unlock() 將鎖解了，這就意味著有可能會有多個 goroutine 來讓條件發生變化。那么，當前 goroutine 是不能保證在 runtime_notifyListWait 返回后，條件就一定是真的，因此需要循環判斷條件。正確的 Wait 使用姿勢如下：

1//    c.L.Lock() 
2//    for !condition() { 
3//        c.Wait() 
4//    } 
5//    ... make use of condition ... 
6//    c.L.Unlock() 

Cond.Signal 方法

1func (c *Cond) Signal() { 
2   c.checker.check() 
3   runtime_notifyListNotifyOne(&c.notify) 
4} 

runtime_notifyListNotifyOne 的詳細實現在 runtime/sema.go 的 notifyListNotifyOne，它的目的就是通知 waiter 取票。具體操作是：如果在上一次通知取票之后沒有新的 waiter 取票者，那么該函數會直接返回。否則，它會將取票號 +1，并通知喚醒等待取票的 waiter。

需要注意的是，調用 Signal 方法時，并不需要持有 c.L 鎖。

Cond.Broadcast 方法

1func (c *Cond) Broadcast() { 
2   c.checker.check() 
3   runtime_notifyListNotifyAll(&c.notify) 
4} 

runtime_notifyListNotifyAll 的詳細實現在 runtime/sema.go 的 notifyListNotifyAll，它會通知喚醒所有的 waiter，并將 notify 值置為 和 wait 值相等。調用 Broadcast 方法時，也不需要持有 c.L 鎖。

討論

在 $GOPATH/src/sync/cond.go 下，我們可以發現其代碼量非常之少，但它呈現的只是核心邏輯，其實現細節位于 runtime/sema.go 之中，依賴的是 runtime 層的調度原語，對細節感興趣的讀者可以深入學習。

問題來了，為什么在日常開發中，我們很少會使用到 sync.Cond ?

無效喚醒

前文中我們提到，使用 Cond.Wait 正確姿勢如下

1    c.L.Lock() 
2    for !condition() { 
3        c.Wait() 
4    } 
5    ... make use of condition ... 
6    c.L.Unlock() 

以文章開頭的例子而言，如果在每次調用 Put 方法時，使用 Broadcast 方法喚醒所有的 waiter，那么很大概率上被喚醒的 waiter 醒來發現條件并不滿足，又會重新進入等待。盡管是調用 Signal 方法喚醒指定的 waiter，但是它也不能保證喚醒的 waiter 條件一定滿足。因此，在實際的使用中，我們需要盡量保證喚醒操作是有效地，為了做到這點，代碼的復雜度難免會增加。

• 饑餓問題

還是以文章開頭例子為例，如果同時有多個 goroutine 執行 GetMany(3) 和 GetMany(3000)，執行 GetMany(3) 與執行 GetMany(3000) 的 goroutine 被喚醒的概率是一樣的，但是由于 GetMany(3) 只需要 3個數據就能滿足條件，那么如果一直存在 GetMany(3) 的 goroutine，執行 GetMany(3000) 的 goroutine 將永遠拿不到數據，一直被無效喚醒。

• 不能響應其他事件

條件變量的意義在于讓 goroutine 等待某種條件發生時進入睡眠狀態。但是這會讓 goroutine 在等待條件時，可能會錯過一些需要注意的其他事件。例如，調用 Cond.Wait 的函數中包含了 context 上下文，當 context 傳來取消信號時，它并不能像我們期望的一樣，獲取到取消信號并退出。Cond 的使用，讓我們不能同時選擇(select)條件和其他事件。

• 可替代性

通過對 sync.Cond 幾個對外方法的分析，我們不難看到，它的使用場景是可以被 channel 所代替的，但是這也會增加代碼的復雜性。上文中的例子，可以使用 channel 改寫如下。

 1type Item = int 
 2 
 3type waiter struct { 
 4    n int 
 5    c chan []Item 
 6} 
 7 
 8type state struct { 
 9    items []Item 
10    wait  []waiter 
11} 
12 
13type Queue struct { 
14    s chan state 
15} 
16 
17func NewQueue() *Queue { 
18    s := make(chan state, 1) 
19    s <- state{} 
20    return &Queue{s} 
21} 
22 
23func (q *Queue) Put(item Item) { 
24    s := <-q.s 
25    s.items = append(s.items, item) 
26    for len(s.wait) > 0 { 
27        w := s.wait[0] 
28        if len(s.items) < w.n { 
29            break 
30        } 
31        w.c <- s.items[:w.n:w.n] 
32        s.items = s.items[w.n:] 
33        s.wait = s.wait[1:] 
34    } 
35    q.s <- s 
36} 
37 
38func (q *Queue) GetMany(n int) []Item { 
39    s := <-q.s 
40    if len(s.wait) == 0 && len(s.items) >= n { 
41        items := s.items[:n:n] 
42        s.items = s.items[n:] 
43        q.s <- s 
44        return items 
45    } 
46 
47    c := make(chan []Item) 
48    s.wait = append(s.wait, waiter{n, c}) 
49    q.s <- s 
50 
51    return <-c 
52} 

最后，雖然在上文的討論中都是列出的 sync.Cond 潛在問題，但是如果開發者能夠在使用中考慮到以上的幾點問題，對于監視器模型的實現而言，在代碼的語義邏輯上，sync.Cond 的使用會比 channel 的模式更易理解和維護。記住一點，通俗易懂的代碼模型總是比深奧的炫技要接地氣。