在 go 的標(biāo)準(zhǔn)庫(kù)中，提供了 sync.Cond 這個(gè)并發(fā)原語(yǔ)，讓我們可以實(shí)現(xiàn)多個(gè) goroutine 等待某一條件滿(mǎn)足之后再繼續(xù)執(zhí)行。 它需要配合 sync.Mutex 一起使用，因?yàn)?Cond 的 Wait 方法需要在 Mutex 的保護(hù)下才能正常工作。 對(duì)于條件變量，可能大多數(shù)人只是知道它的存在，但是用到它的估計(jì)寥寥無(wú)幾，因?yàn)楹芏嗖l(fā)場(chǎng)景的處理都能使用 chan 來(lái)實(shí)現(xiàn)， 而且 chan 的使用也更加簡(jiǎn)單。 但是在某些場(chǎng)景下，Cond 可能是最好的選擇，本文就來(lái)探討一下 Cond 的使用場(chǎng)景，基本用法，以及它的實(shí)現(xiàn)原理。

sync.Cond 是什么

sync.Cond 表示的是條件變量，它是一種同步機(jī)制，用來(lái)協(xié)調(diào)多個(gè) goroutine 之間的同步，當(dāng)共享資源的狀態(tài)發(fā)生變化的時(shí)候， 可以通過(guò)條件變量來(lái)通知所有等待的 goroutine 去重新獲取共享資源。

適用場(chǎng)景

在實(shí)際使用中，我們可能會(huì)有多個(gè) goroutine 在執(zhí)行的過(guò)程中，由于某一條件不滿(mǎn)足而阻塞的情況。 這個(gè)時(shí)候，我們就可以使用條件變量來(lái)實(shí)現(xiàn) goroutine 之間的同步。比如，我們有一個(gè) goroutine 用來(lái)獲取數(shù)據(jù)， 但是可能會(huì)比較耗時(shí)，這個(gè)時(shí)候，我們就可以使用條件變量來(lái)實(shí)現(xiàn) goroutine 之間的同步， 當(dāng)數(shù)據(jù)準(zhǔn)備好之后，就可以通過(guò)條件變量來(lái)通知所有等待的 goroutine 去重新獲取共享資源。

sync.Cond 條件變量用來(lái)協(xié)調(diào)想要訪問(wèn)共享資源的那些 goroutine，當(dāng)共享資源的狀態(tài)發(fā)生變化的時(shí)候， 它可以用來(lái)通知所有等待的 goroutine 去重新獲取共享資源。

sync.Cond 的基本用法

sync.Cond 的基本用法非常簡(jiǎn)單，我們只需要通過(guò) sync.NewCond 方法來(lái)創(chuàng)建一個(gè) Cond 實(shí)例， 然后通過(guò) Wait 方法來(lái)等待條件滿(mǎn)足，通過(guò) Signal 或者 Broadcast 方法來(lái)通知所有等待的 goroutine 去重新獲取共享資源。

NewCond 創(chuàng)建實(shí)例

sync.NewCond 方法用來(lái)創(chuàng)建一個(gè) Cond 實(shí)例，它的參數(shù)是一個(gè) Locker 接口，我們可以傳入一個(gè) Mutex 或者 RWMutex 實(shí)例。 這個(gè)條件變量的 Locker 接口就是用來(lái)保護(hù)共享資源的。

Wait 等待條件滿(mǎn)足

Wait 方法用來(lái)等待條件滿(mǎn)足，它會(huì)先釋放 Cond 的鎖(Cond.L)，然后阻塞當(dāng)前 goroutine(實(shí)際調(diào)用的是 goparkunlock)，直到被 Signal 或者 Broadcast 喚醒。

它做了如下幾件事情：

• 釋放 Cond 的鎖(Cond.L)，然后阻塞當(dāng)前 goroutine。（所以，使用之前需要先鎖定）
• 被 Signal 或者 Broadcast 喚醒之后，會(huì)重新獲取 Cond 的鎖(Cond.L)。
• 之后，就返回到 goroutine 阻塞的地方繼續(xù)執(zhí)行。

Signal 通知一個(gè)等待的 goroutine

Signal 方法用來(lái)通知一個(gè)等待的 goroutine，它會(huì)喚醒一個(gè)等待的 goroutine，然后繼續(xù)執(zhí)行當(dāng)前 goroutine。 如果沒(méi)有等待的 goroutine，則不會(huì)有任何操作。

Broadcast 通知所有等待的 goroutine

Broadcast 方法用來(lái)通知所有等待的 goroutine，它會(huì)喚醒所有等待的 goroutine，然后繼續(xù)執(zhí)行當(dāng)前 goroutine。 如果沒(méi)有等待的 goroutine，則不會(huì)有任何操作。

sync.Cond 使用實(shí)例

下面我們通過(guò)一個(gè)實(shí)例來(lái)看一下 sync.Cond 的使用方法。

package cond

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

var done bool
var data string

func write(c *sync.Cond) {
   fmt.Println("writing.")
   // 讓 reader 先獲取鎖，模擬條件不滿(mǎn)足然后 wait 的情況
   time.Sleep(time.Millisecond * 10)
   c.L.Lock()
   // 模擬耗時(shí)的寫(xiě)操作
   time.Sleep(time.Millisecond * 50)
   data = "hello world"
   done = true
   fmt.Println("writing done.")
   c.L.Unlock()
   c.Broadcast()
}

func read(c *sync.Cond) {
   fmt.Println("reading")
   c.L.Lock()
   for !done {
      fmt.Println("reader wait.")
      c.Wait()
   }
   fmt.Println("read done.")
   fmt.Println("data:", data)
   defer c.L.Unlock()
}

func TestCond(t *testing.T) {
   var c = sync.NewCond(&sync.Mutex{})

   go read(c)  // 讀操作
   go read(c)  // 讀操作
   go write(c) // 寫(xiě)操作

   time.Sleep(time.Millisecond * 100) // 等待操作完成
}

輸出：

reading
reader wait. // 還沒(méi)獲取完數(shù)據(jù)，需要等待
writing.
reading
reader wait.
writing done. // 獲取完數(shù)據(jù)了，通知所有等待的 reader
read done. // 讀取到數(shù)據(jù)了
data: hello world // 輸出讀取到的數(shù)據(jù)
read done.
data: hello world

這個(gè)例子可以粗略地用下圖來(lái)表示：

說(shuō)明：

• read1 和 reader2 表示兩個(gè) goroutine，它們都會(huì)調(diào)用 read 函數(shù)。
• 在 done 為 false 的時(shí)候，reader1 和 reader2 都會(huì)調(diào)用 c.Wait() 函數(shù)，然后阻塞等待。
• write 表示一個(gè) goroutine，它會(huì)調(diào)用 write 函數(shù)。
• 在 write 函數(shù)中，獲取完數(shù)據(jù)之后，會(huì)將 done 設(shè)置為 true，然后調(diào)用 c.Broadcast() 函數(shù)，通知所有等待的 reader 去重新獲取共享資源。
• reader1 和 reader2 在解除阻塞狀態(tài)后，都會(huì)重新獲取共享資源，然后輸出讀取到的數(shù)據(jù)。

在這個(gè)例子中，done 的功能是標(biāo)記，用來(lái)表示共享資源是否已經(jīng)獲取完畢，如果沒(méi)有獲取完畢，那么 reader 就會(huì)阻塞等待。

為什么要用 sync.Cond

在文章開(kāi)頭，我們說(shuō)了，很多并發(fā)編程的問(wèn)題都可以通過(guò) channel 來(lái)解決。 同樣的，在上面提到的 sync.Cond 的使用場(chǎng)景，使用 channel 也是可以實(shí)現(xiàn)的， 我們只要 close(ch) 來(lái)關(guān)閉 channel 就可以實(shí)現(xiàn)通知多個(gè)等待的協(xié)程了。

那么為什么還要用 sync.Cond 呢？ 主要原因是，sync.Cond 可以重復(fù)地進(jìn)行 Wait() 和 Signal()、Broadcast() 操作， 但是，如果想通過(guò)關(guān)閉 chan 來(lái)實(shí)現(xiàn)這個(gè)功能的話，那就只能通知一次了。 因?yàn)?channel 只能關(guān)閉一次，關(guān)閉一個(gè)已經(jīng)關(guān)閉的 channel 會(huì)導(dǎo)致程序 panic。

使用 channel 的另外一種方式是，記錄 reader 的數(shù)量，然后通過(guò)往 channel 中發(fā)送多次數(shù)據(jù)來(lái)實(shí)現(xiàn)通知多個(gè) reader。 但是這樣一來(lái)代碼就會(huì)復(fù)雜很多，從另一個(gè)角度說(shuō)，出錯(cuò)的概率大了很多。

close channel 廣播實(shí)例

下面的例子模擬了使用 close(chan) 來(lái)實(shí)現(xiàn) sync.Cond 中那種廣播功能，但是只能通知一次。

package close_chan

import (
   "fmt"
   "testing"
   "time"
)

var data string

func read(c <-chan struct{}) {
   fmt.Println("reading.")

   // 從 chan 接收數(shù)據(jù)，如果 chan 中沒(méi)有數(shù)據(jù)，會(huì)阻塞。
   // 如果能接收到數(shù)據(jù)，或者 chan 被關(guān)閉，會(huì)解除阻塞狀態(tài)。
   <-c

   fmt.Println("data:", data)
}

func write(c chan struct{}) {
   fmt.Println("writing.")
   // 模擬耗時(shí)的寫(xiě)操作
   time.Sleep(time.Millisecond * 10)
   data = "hello world"
   fmt.Println("write done.")

   // 關(guān)閉 chan 的時(shí)候，會(huì)通知所有的 reader
   // 所有等待從 chan 接收數(shù)據(jù)的 goroutine 都會(huì)被喚醒
   close(c)
}

func TestCloseChan(t *testing.T) {
   ch := make(chan struct{})

   go read(ch)
   go read(ch)
   go write(ch)

   // 不能關(guān)閉已經(jīng)關(guān)閉的 chan
   time.Sleep(time.Millisecond * 20)
   // panic: close of closed channel
   // 下面這行代碼會(huì)導(dǎo)致 panic
   //go write(ch)

   time.Sleep(time.Millisecond * 100)
}

輸出：

writing.
reading. // 會(huì)阻塞直到寫(xiě)完
reading. // 會(huì)阻塞直到寫(xiě)完
write done. // 寫(xiě)完之后，才能讀
data: hello world
data: hello world

上面例子的 write 不能多次調(diào)用，否則會(huì)導(dǎo)致 panic。

sync.Cond 基本原理

go 的 sync.Cond 中維護(hù)了一個(gè)鏈表，這個(gè)鏈表記錄了所有阻塞的 goroutine，也就是由于調(diào)用了 Wait 而阻塞的 goroutine。 而 Signal 和 Broadcast 方法就是用來(lái)喚醒這個(gè)鏈表中的 goroutine 的。 Signal 方法只會(huì)喚醒鏈表中的第一個(gè) goroutine，而 Broadcast 方法會(huì)喚醒鏈表中的所有 goroutine。

下圖是 Signal 方法的效果，可以看到，Signal 方法只會(huì)喚醒鏈表中的第一個(gè) goroutine：

說(shuō)明：

• notifyList 是 sync.Cond 中維護(hù)的一個(gè)鏈表，這個(gè)鏈表記錄了所有阻塞的 goroutine。
• head 是鏈表的頭節(jié)點(diǎn)，tail 是鏈表的尾節(jié)點(diǎn)。
• Signal 方法只會(huì)喚醒鏈表中的第一個(gè) goroutine。

而 Broadcast 方法會(huì)喚醒 notifyList 中的所有 goroutine。

sync.Cond 的設(shè)計(jì)與實(shí)現(xiàn)

最后，我們來(lái)看一下 sync.Cond 的設(shè)計(jì)與實(shí)現(xiàn)。

sync.Cond 模型

sync.Cond 的模型如下所示：

type Cond struct {
   noCopy noCopy

   // L is held while observing or changing the condition
   L Locker // L 在觀察或改變條件時(shí)被持有

   notify  notifyList
   checker copyChecker
}

屬性說(shuō)明：

• noCopy 是一個(gè)空結(jié)構(gòu)體，用來(lái)檢查 sync.Cond 是否被復(fù)制。（在編譯前通過(guò) go vet 命令來(lái)檢查）
• L 是一個(gè) Locker 接口，用來(lái)保護(hù)條件變量。
• notify 是一個(gè) notifyList 類(lèi)型，用來(lái)記錄所有阻塞的 goroutine。
• checker 是一個(gè) copyChecker 類(lèi)型，用來(lái)檢查 sync.Cond 是否被復(fù)制。（如果在運(yùn)行時(shí)被復(fù)制，會(huì)導(dǎo)致 panic）

notifyList 結(jié)構(gòu)體

notifyList 是 sync.Cond 中維護(hù)的一個(gè)鏈表，這個(gè)鏈表記錄了所有因?yàn)楣蚕碣Y源還沒(méi)準(zhǔn)備好而阻塞的 goroutine。它的定義如下所示：

type notifyList struct {
   wait atomic.Uint32
   notify uint32

   // 阻塞的 waiter 名單。
   lock mutex // 鎖
   head *sudog // 阻塞的 goroutine 鏈表（鏈表頭）
   tail *sudog // 阻塞的 goroutine 鏈表（鏈表尾）
}

屬性說(shuō)明：

• wait 是下一個(gè) waiter 的編號(hào)。它在鎖外自動(dòng)遞增。
• notify 是下一個(gè)要通知的 waiter 的編號(hào)。它可以在鎖外讀取，但只能在持有鎖的情況下寫(xiě)入。
• lock 是一個(gè) mutex 類(lèi)型，用來(lái)保護(hù) notifyList。
• head 是一個(gè) sudog 類(lèi)型，用來(lái)記錄阻塞的 goroutine 鏈表的頭節(jié)點(diǎn)。
• tail 是一個(gè) sudog 類(lèi)型，用來(lái)記錄阻塞的 goroutine 鏈表的尾節(jié)點(diǎn)。

notifyList 的方法說(shuō)明：

notifyList 中包含了幾個(gè)操作阻塞的 goroutine 鏈表的方法。

• notifyListAdd 方法將 waiter 的編號(hào)加 1。
• notifyListWait 方法將當(dāng)前的 goroutine 加入到 notifyList 中。（也就是將當(dāng)前協(xié)程掛起）
• notifyListNotifyOne 方法將 notifyList 中的第一個(gè) goroutine 喚醒。
• notifyListNotifyAll 方法將 notifyList 中的所有 goroutine 喚醒。
• notifyListCheck 方法檢查 notifyList 的大小是否正確。

sync.Cond 的方法

notifyList 就不細(xì)說(shuō)了，本文重點(diǎn)講解一下 sync.Cond 的實(shí)現(xiàn)。

Wait 方法

Wait 方法用在當(dāng)條件不滿(mǎn)足的時(shí)候，將當(dāng)前運(yùn)行的協(xié)程掛起。

func (c *Cond) Wait() {
   // 檢查是否被復(fù)制
   c.checker.check()
   // 更新 notifyList 中需要等待的 waiter 的數(shù)量
   // 返回當(dāng)前需要插入 notifyList 的編號(hào)
   t := runtime_notifyListAdd(&c.notify)
   // 解鎖
   c.L.Unlock()
   // 掛起當(dāng)前 g，直到被喚醒
   runtime_notifyListWait(&c.notify, t)
   // 喚醒之后，重新加鎖。
   // 因?yàn)樽枞敖怄i了。
   c.L.Lock()
}

對(duì)于 Wait 方法，我們需要注意的是，使用之前，我們需要先調(diào)用 L.Lock() 方法加鎖，然后再調(diào)用 Wait 方法，否則會(huì)報(bào)錯(cuò)。

文檔里面的例子：

c.L.Lock()
for !condition() {
    c.Wait()
}
// ...使用條件...
// 這里是我們?cè)跅l件滿(mǎn)足之后，需要執(zhí)行的代碼。
c.L.Unlock()

好了，問(wèn)題來(lái)了，調(diào)用 Wait 方法之前為什么要先加鎖呢？

這是因?yàn)樵谖覀兪褂霉蚕碣Y源的時(shí)候，可能一些代碼是互斥的，所以我們需要加鎖。 這樣我們就可以保證在我們使用共享資源的時(shí)候，不會(huì)被其他協(xié)程修改。 但是如果因?yàn)闂l件不滿(mǎn)足，我們需要等待的話，我們不可能在持有鎖的情況下等待， 因?yàn)樵谛薷臈l件的時(shí)候，可能也需要加鎖，這樣就會(huì)造成死鎖。

另外一個(gè)問(wèn)題是，為什么要使用 for 來(lái)檢查條件是否滿(mǎn)足，而不是使用 if 呢？

這是因?yàn)樵谖覀冋{(diào)用 Wait 方法之后，可能會(huì)有其他協(xié)程喚醒我們，但是條件并沒(méi)有滿(mǎn)足， 這個(gè)時(shí)候依然是需要繼續(xù) Wait 的。

Signal 方法

Signal 方法用在當(dāng)條件滿(mǎn)足的時(shí)候，將 notifyList 中的第一個(gè) goroutine 喚醒。

func (c *Cond) Signal() {
   // 檢查 sync.Cond 是否被復(fù)制了
   c.checker.check()
   // 喚醒 notifyList 中的第一個(gè) goroutine
   runtime_notifyListNotifyOne(&c.notify)
}

Broadcast 方法

Broadcast 方法用在當(dāng)條件滿(mǎn)足的時(shí)候，將 notifyList 中的所有 goroutine 喚醒。

func (c *Cond) Broadcast() {
   // 檢查 sync.Cond 是否被復(fù)制了
   c.checker.check()
   // 喚醒 notifyList 中的所有 goroutine
   runtime_notifyListNotifyAll(&c.notify)
}

copyChecker 結(jié)構(gòu)體

copyChecker 結(jié)構(gòu)體用來(lái)檢查 sync.Cond 是否被復(fù)制。它實(shí)際上只是一個(gè) uintptr 類(lèi)型的值。

type copyChecker uintptr

// check 方法檢查 copyChecker 是否被復(fù)制了。
func (c *copyChecker) check() {
   if uintptr(*c) != uintptr(unsafe.Pointer(c)) &&
      !atomic.CompareAndSwapUintptr((*uintptr)(c), 0, uintptr(unsafe.Pointer(c))) &&
      uintptr(*c) != uintptr(unsafe.Pointer(c)) {
      panic("sync.Cond is copied")
   }
}

copyChecker 的值只有兩種可能：

• 0，表示還沒(méi)有調(diào)用過(guò) Wait, Signal 或 Broadcast 方法。
• uintptr(unsafe.Pointer(&copyChecker))，表示已經(jīng)調(diào)用過(guò) Wait, Signal 或 Broadcast 方法。在這幾個(gè)方法里面會(huì)調(diào)用 check 方法，所以 copyChecker 的值會(huì)被修改。

所以如果 copyChecker 的值不是 0，也不是 uintptr(unsafe.Pointer(&copyChecker))（也就是最初的 copyChecker 的內(nèi)存地址），則表示 copyChecker 被復(fù)制了。

需要注意的是，這個(gè)方法在調(diào)用 CompareAndSwapUintptr 還會(huì)檢查一下，這是因?yàn)橛锌赡軙?huì)并發(fā)調(diào)用 CompareAndSwapUintptr， 如果另外一個(gè)協(xié)程調(diào)用了 CompareAndSwapUintptr 并且成功了，那么當(dāng)前協(xié)程的這個(gè) CompareAndSwapUintptr 調(diào)用會(huì)返回 false， 這個(gè)時(shí)候就需要檢查是否是因?yàn)榱硗庖粋€(gè)協(xié)程調(diào)用了 CompareAndSwapUintptr 而導(dǎo)致的，如果是的話，就不會(huì) panic。

為什么 sync.Cond 不能被復(fù)制

從上一小節(jié)中我們可以看到，sync.Cond 其實(shí)是不允許被復(fù)制的，但是如果是在調(diào)用 Wait, Signal 或 Broadcast 方法之前復(fù)制，那倒是沒(méi)關(guān)系。

這是因?yàn)?sync.Cond 中維護(hù)了一個(gè)阻塞的 goroutine 列表。如果 sync.Cond 被復(fù)制了，那么這個(gè)列表就會(huì)被復(fù)制，這樣就會(huì)導(dǎo)致兩個(gè) sync.Cond 都包含了這個(gè)列表；但是我們喚醒的時(shí)候，只會(huì)有其中一個(gè) sync.Cond 被喚醒，另外一個(gè) sync.Cond 就會(huì)一直阻塞。 所以 go 直接從語(yǔ)言層面限制了這種情況，不允許 sync.Cond 被復(fù)制。

總結(jié)

sync.Cond 是一個(gè)條件變量，它可以用來(lái)協(xié)調(diào)多個(gè) goroutine 之間的同步，當(dāng)條件滿(mǎn)足的時(shí)候，去通知那些因?yàn)闂l件不滿(mǎn)足被阻塞的 goroutine 繼續(xù)執(zhí)行。

sync.Cond 的接口比較簡(jiǎn)單，只有 Wait, Signal 和 Broadcast 三個(gè)方法。

• Wait 方法用來(lái)阻塞當(dāng)前 goroutine，直到條件滿(mǎn)足。調(diào)用 Wait 方法之前，需要先調(diào)用 L.Lock 方法加鎖。
• Signal 方法用來(lái)喚醒 notifyList 中的第一個(gè) goroutine。
• Broadcast 方法用來(lái)喚醒 notifyList 中的所有 goroutine。

sync.Cond 的實(shí)現(xiàn)也比較簡(jiǎn)單，它的核心就是 notifyList，它是一個(gè)鏈表，用來(lái)保存所有因?yàn)闂l件不滿(mǎn)足而被阻塞的 goroutine。

用關(guān)閉 channel 的方式也可以實(shí)現(xiàn)類(lèi)似的廣播功能，但是有個(gè)問(wèn)題是 channel 不能被重復(fù)關(guān)閉，所以這種方式無(wú)法被多次使用。也就是說(shuō)使用這種方式無(wú)法多次廣播。

使用 channel 發(fā)送通知的方式也是可以的，但是這樣實(shí)現(xiàn)起來(lái)就復(fù)雜很多了，就更容易出錯(cuò)了。

sync.Cond 中使用 copyChecker 來(lái)檢查 sync.Cond 是否被復(fù)制，如果被復(fù)制了，就會(huì) panic。需要注意的是，這里的復(fù)制是指調(diào)用了 Wait，Signal 或 Broadcast 方法之后，sync.Cond 被復(fù)制了。在調(diào)用這幾個(gè)方法之前進(jìn)行復(fù)制是沒(méi)有影響的。

以上就是一文帶你深入理解Go語(yǔ)言中的sync.Cond的詳細(xì)內(nèi)容，更多關(guān)于Go語(yǔ)言 sync.Cond的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論