深入學(xué)習(xí)Golang并發(fā)編程必備利器之sync.Cond類型

更新時(shí)間：2023年05月04日 09:15:48 作者：金刀大菜牙

Go?語(yǔ)言的?sync?包提供了一系列同步原語(yǔ)，其中?sync.Cond?就是其中之一。本文將深入探討?sync.Cond?的實(shí)現(xiàn)原理和使用方法，幫助大家更好地理解和應(yīng)用?sync.Cond，需要的可以參考一下

Go 語(yǔ)言的 sync 包提供了一系列同步原語(yǔ)，其中 sync.Cond 就是其中之一。sync.Cond 的作用是在多個(gè) goroutine 之間進(jìn)行條件變量的同步。本文將深入探討 sync.Cond 的實(shí)現(xiàn)原理和使用方法，幫助大家更好地理解和應(yīng)用 sync.Cond。

1. sync.Cond 的基本概念

1.1 條件變量

條件變量是一種同步機(jī)制，用于在多個(gè) goroutine 之間進(jìn)行同步。條件變量通常是和互斥鎖一起使用的，用于等待某個(gè)條件的出現(xiàn)。

在 Go 語(yǔ)言中，條件變量由 sync.Cond 類型實(shí)現(xiàn)。它提供了兩個(gè)主要的方法：Wait 和 Signal/Broadcast。Wait 方法用于等待條件變量的出現(xiàn)，Signal/Broadcast 方法用于通知等待中的 goroutine。

1.2 互斥鎖

互斥鎖是一種用于控制對(duì)共享資源訪問的同步機(jī)制。它能夠保證同一時(shí)刻只有一個(gè) goroutine 能夠訪問共享資源。

在 Go 語(yǔ)言中，互斥鎖由 sync.Mutex 類型實(shí)現(xiàn)。它提供了兩個(gè)主要的方法：Lock 和 Unlock。Lock 方法用于加鎖，保證同一時(shí)刻只有一個(gè) goroutine 能夠訪問共享資源；Unlock 方法用于解鎖，允許其他 goroutine 訪問共享資源。

1.3 條件變量的實(shí)現(xiàn)原理

條件變量的實(shí)現(xiàn)原理基于互斥鎖和 goroutine 隊(duì)列。

假設(shè)有一個(gè)條件變量 cond，初始時(shí)它沒有被觸發(fā)。當(dāng)一個(gè) goroutine 調(diào)用 cond.Wait() 方法時(shí)，它會(huì)加鎖并將自己加入到 cond 的 goroutine 隊(duì)列中。接著，它會(huì)解鎖并進(jìn)入睡眠狀態(tài)，等待被喚醒。

當(dāng)另一個(gè) goroutine 調(diào)用 cond.Signal() 或者 cond.Broadcast() 方法時(shí)，它會(huì)重新加鎖，并從 cond 的 goroutine 隊(duì)列中選擇一個(gè) goroutine 喚醒。被喚醒的 goroutine 會(huì)重新加鎖，然后繼續(xù)執(zhí)行。

需要注意的是，被喚醒的 goroutine 并不會(huì)立即執(zhí)行，它會(huì)等待重新獲得鎖之后才會(huì)繼續(xù)執(zhí)行。

2. sync.Cond 的基本用法

2.1 創(chuàng)建 sync.Cond 對(duì)象

sync.Cond 對(duì)象需要依賴一個(gè) sync.Mutex 或 sync.RWMutex 對(duì)象來進(jìn)行同步和互斥操作。我們可以使用 sync.NewCond 方法來創(chuàng)建一個(gè)新的 sync.Cond 對(duì)象，該方法接受一個(gè) Mutex 或 RWMutex 對(duì)象作為參數(shù)，返回一個(gè)對(duì)應(yīng)的條件變量對(duì)象。

 package main
 ?
 import (
     "fmt"
     "sync"
 )
 ?
 func main() {
     var mu sync.Mutex
     cond := sync.NewCond(&mu)
 ?
     // ...
 }

2.2 等待條件變量

sync.Cond 提供了 Wait 方法來等待條件變量的信號(hào)。Wait 方法需要在持有 Mutex 或 RWMutex 的情況下進(jìn)行調(diào)用，否則會(huì)拋出 panic 異常。

 func (c *Cond) Wait()

Wait 方法將當(dāng)前 goroutine 暫停，等待條件變量的信號(hào)。在等待過程中，Mutex 或 RWMutex 將被釋放，其他 goroutine 可以獲取鎖并修改共享變量，但是當(dāng)前 goroutine 仍然保持在等待隊(duì)列中，直到收到喚醒信號(hào)。當(dāng) Wait 方法返回時(shí)，Mutex 或 RWMutex 會(huì)自動(dòng)重新被鎖定。

下面是一個(gè)簡(jiǎn)單的示例程序，使用 sync.Cond 實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的條件等待機(jī)制：

 package main
 ?
 import (
     "fmt"
     "sync"
     "time"
 )
 ?
 func main() {
     var mu sync.Mutex
     cond := sync.NewCond(&mu)
     var ready bool
 ?
     // 模擬一個(gè)耗時(shí)的初始化操作
     go func() {
         time.Sleep(2 * time.Second)
         mu.Lock()
         ready = true
         cond.Signal() // 喚醒等待的 goroutine
         mu.Unlock()
     }()
 ?
     mu.Lock()
     for !ready {
         cond.Wait() // 等待初始化完成信號(hào)
     }
     fmt.Println("Initialization completed")
     mu.Unlock()
 }

上面的示例程序中，我們通過 sync.Cond 實(shí)現(xiàn)了一種等待初始化完成的機(jī)制。在初始化完成前，主 goroutine 會(huì)等待條件變量的信號(hào)，當(dāng)子 goroutine 完成初始化后，會(huì)通過 Signal 方法發(fā)送喚醒信號(hào)，使得主 goroutine 繼續(xù)執(zhí)行。

2.3 喚醒等待的 goroutine

sync.Cond 提供了兩種方式來喚醒等待的 goroutine：Signal 和 Broadcast。

2.3.1 Signal 方法

Signal 方法用于喚醒等待隊(duì)列中的一個(gè) goroutine，使其繼續(xù)執(zhí)行。在調(diào)用 Signal 方法之前，必須先獲得 Mutex 或 RWMutex 的鎖。

 func (c *Cond) Signal()

Signal 方法會(huì)選擇等待隊(duì)列中的一個(gè) goroutine 并喚醒它，如果沒有等待的 goroutine，那么 Signal 方法不會(huì)產(chǎn)生任何效果。

下面是一個(gè)示例程序，演示了如何使用 Signal 方法喚醒等待的 goroutine：

 package main
 ?
 import (
     "fmt"
     "sync"
     "time"
 )
 ?
 func main() {
     var mu sync.Mutex
     cond := sync.NewCond(&mu)
     var ready bool
 ?
     // 模擬一個(gè)耗時(shí)的初始化操作
     go func() {
         time.Sleep(2 * time.Second)
         mu.Lock()
         ready = true
         cond.Signal() // 喚醒等待的 goroutine
         mu.Unlock()
     }()
 ?
     mu.Lock()
     for !ready {
         cond.Wait() // 等待初始化完成信號(hào)
     }
     fmt.Println("Initialization completed")
     mu.Unlock()
 }

在上面的示例程序中，我們通過調(diào)用 cond.Signal() 方法來喚醒等待的 goroutine。

2.3.2 Broadcast 方法

Broadcast 方法用于喚醒等待隊(duì)列中的所有 goroutine，使它們繼續(xù)執(zhí)行。在調(diào)用 Broadcast 方法之前，必須先獲得 Mutex 或 RWMutex 的鎖。

 func (c *Cond) Broadcast()

Broadcast 方法會(huì)喚醒等待隊(duì)列中的所有 goroutine，如果沒有等待的 goroutine，那么 Broadcast 方法不會(huì)產(chǎn)生任何效果。

下面是一個(gè)示例程序，演示了如何使用 Broadcast 方法喚醒等待的 goroutine：

 package main
 ?
 import (
     "fmt"
     "sync"
     "time"
 )
 ?
 func main() {
     var mu sync.Mutex
     cond := sync.NewCond(&mu)
     var ready bool
 ?
     // 模擬一個(gè)耗時(shí)的初始化操作
     go func() {
         time.Sleep(2 * time.Second)
         mu.Lock()
         ready = true
         cond.Broadcast() // 喚醒等待的所有 goroutine
         mu.Unlock()
     }()
 ?
     mu.Lock()
     for !ready {
         cond.Wait() // 等待初始化完成信號(hào)
     }
     fmt.Println("Initialization completed")
     mu.Unlock()
 }

在上面的示例程序中，我們通過調(diào)用 cond.Broadcast() 方法來喚醒等待的 goroutine。

3. sync.Cond 的內(nèi)部實(shí)現(xiàn)原理

sync.Cond 的內(nèi)部實(shí)現(xiàn)依賴于一個(gè)等待隊(duì)列，它維護(hù)了等待條件變量的 goroutine 的列表，其中每個(gè) goroutine 都有一個(gè)阻塞的狀態(tài)。當(dāng)條件變量被發(fā)出信號(hào)時(shí)，等待隊(duì)列中的一個(gè) goroutine 將被喚醒，并從 Wait 方法中返回，同時(shí)將重新獲得 Mutex 的鎖。

下面是 sync.Cond 內(nèi)部的等待隊(duì)列結(jié)構(gòu)體定義：

 type wait struct {
     // 等待隊(duì)列中的 goroutine
     // goroutine 在 cond.Wait() 中被加入隊(duì)列，在 cond.Signal() 或 cond.Broadcast() 中被喚醒
     // 由于隊(duì)列是單向鏈表，因此需要保存 next 指針指向下一個(gè)元素
     // 當(dāng) goroutine 被喚醒時(shí)，會(huì)將 wait.done 設(shè)置為 true，并喚醒 wait.cond.L 上阻塞的 goroutine
     // goroutine 從 Wait() 方法中返回時(shí)，會(huì)將 wait.done 設(shè)置為 true
     // wait.done 可以保證 goroutine 不會(huì)重復(fù)地從 cond.Wait() 方法中返回
     // wait.done 可以保證 goroutine 在從 cond.Wait() 方法中返回時(shí)，已經(jīng)持有了 Mutex 的鎖
     // wait.done 可以保證 goroutine 在被喚醒之前不會(huì)在 cond.Wait() 方法中被重新加入到隊(duì)列中
     done bool
     // 下一個(gè)等待隊(duì)列元素的指針
     next *wait
     // 條件變量
     cond *Cond
 }

sync.Cond 使用 wait 結(jié)構(gòu)體維護(hù)了一個(gè)等待隊(duì)列，其中每個(gè)元素都代表了一個(gè)等待 goroutine。

wait 結(jié)構(gòu)體中的 done 字段用于保證 goroutine 不會(huì)重復(fù)地從 Wait 方法中返回，next 字段用于鏈接下一個(gè)等待元素。

等待隊(duì)列的頭部和尾部分別使用 wait 結(jié)構(gòu)體的指針 first 和 last 維護(hù)。

 type Cond struct {
     // Mutex 保護(hù) condition 變量和等待隊(duì)列
     L Locker
 ?
     // 等待隊(duì)列的頭部和尾部
     first *wait
     last  *wait
 }

sync.Cond 的 Wait 方法實(shí)現(xiàn)如下：

 func (c *Cond) Wait() {
     // 將當(dāng)前 goroutine 加入到等待隊(duì)列中
     t := new(wait)
     t.cond = c
     c.add(t)
     defer c.remove(t)
 ?
     // 釋放鎖并進(jìn)入阻塞狀態(tài)
     c.L.Unlock()
     for !t.done {
         runtime.Gosched()
     }
     c.L.Lock()
 }

在 Wait 方法中，首先創(chuàng)建一個(gè) wait 結(jié)構(gòu)體 t，并將當(dāng)前 goroutine 加入到等待隊(duì)列中，然后釋放 Mutex 的鎖，并進(jìn)入阻塞狀態(tài)。

在等待隊(duì)列中，goroutine 的狀態(tài)為阻塞，直到被喚醒并從 Wait 方法中返回。

當(dāng)?shù)却臈l件變量滿足時(shí)，喚醒等待隊(duì)列中的 goroutine 的操作由 Signal 和 Broadcast 方法來實(shí)現(xiàn)。

Signal 方法會(huì)喚醒等待隊(duì)列中的一個(gè) goroutine，而 Broadcast 方法會(huì)喚醒所有等待隊(duì)列中的 goroutine。

 func (c *Cond) Signal() {
     if c.first != nil {
         c.first.wake(true)
     }
 }
 ?
 func (c *Cond) Broadcast() {
     for c.first != nil {
         c.first.wake(true)
     }
 }

在 Signal 和 Broadcast 方法中，首先判斷等待隊(duì)列是否為空，如果不為空，則喚醒等待隊(duì)列中的一個(gè)或所有 goroutine，并將它們從阻塞狀態(tài)中解除。下面是 wait 結(jié)構(gòu)體的 wake 方法實(shí)現(xiàn)：

 func (w *wait) wake(done bool) {
     // 標(biāo)記 done 字段并解除阻塞狀態(tài)
     w.done = done
     runtime.NotifyListNotify(&w.cond.L.(*Mutex).notify)
 }

在 wake 方法中，首先將 wait.done 設(shè)置為 true，然后通過調(diào)用 runtime.NotifyListNotify 方法，將等待隊(duì)列中的 goroutine 從阻塞狀態(tài)中解除。

這里需要注意的是，在 sync.Cond 的實(shí)現(xiàn)中，使用了 Mutex 的 notify 字段來實(shí)現(xiàn) goroutine 的喚醒和阻塞。

當(dāng)一個(gè) goroutine 調(diào)用 Wait 方法時(shí)，它會(huì)釋放 Mutex 的鎖，并進(jìn)入阻塞狀態(tài)，同時(shí)將自己加入到 Mutex 的 notify 隊(duì)列中。

當(dāng)一個(gè) goroutine 調(diào)用 Signal 或 Broadcast 方法時(shí)，它會(huì)從 Mutex 的 notify 隊(duì)列中取出一個(gè)或多個(gè) goroutine，并喚醒它們。

這種實(shí)現(xiàn)方式與操作系統(tǒng)的線程調(diào)度機(jī)制類似，可以保證喚醒的 goroutine 在調(diào)用 Wait 方法時(shí)已經(jīng)持有了 Mutex 的鎖，從而避免了死鎖和競(jìng)態(tài)條件等問題。

這里再補(bǔ)充一下 Mutex 的 notify 字段的定義：

 type Mutex struct {
     state int32
     sema  uint32
     waitm uint32
     notify notifyList
 }

notify 字段是一個(gè) notifyList 類型的對(duì)象，它定義如下：

 type notifyList struct {
     wait   uint32 // 等待的 goroutine 的數(shù)量
     notify uint32 // 喚醒的 goroutine 的數(shù)量
     head   *wait  // 等待隊(duì)列的頭部元素
     tail   *wait  // 等待隊(duì)列的尾部元素
 }

notifyList 類型的對(duì)象維護(hù)了一個(gè)等待隊(duì)列和喚醒隊(duì)列，其中等待隊(duì)列用于存放阻塞的 goroutine，喚醒隊(duì)列用于存放將要被喚醒的 goroutine。

notifyList 類型的對(duì)象還維護(hù)了等待隊(duì)列和喚醒隊(duì)列中 goroutine 的數(shù)量。

當(dāng)一個(gè) goroutine 調(diào)用 Wait 方法時(shí)，它會(huì)將自己加入到等待隊(duì)列中，并且將 Mutex 的 waitm 字段加一。

當(dāng)一個(gè) goroutine 調(diào)用 Signal 或 Broadcast 方法時(shí)，它會(huì)從等待隊(duì)列中取出一個(gè)或多個(gè) goroutine，并將它們加入到喚醒隊(duì)列中。

當(dāng)一個(gè) goroutine 調(diào)用 Unlock 方法時(shí)，它會(huì)判斷喚醒隊(duì)列中是否有 goroutine 需要喚醒，并將 Mutex 的 sema 字段加一，從而使得下一個(gè) goroutine 獲得鎖。

4. sync.Cond 的使用方法

sync.Cond 的使用方法通常包括以下步驟：

1.定義互斥鎖和條件變量。

 var mutex sync.Mutex
 var cond = sync.NewCond(&mutex)

2.在生產(chǎn)者和消費(fèi)者之間使用互斥鎖和條件變量進(jìn)行同步。

 package main
 ?
 import (
     "fmt"
     "math/rand"
     "sync"
     "time"
 )
 ?
 type Queue struct {
     items []int
     size  int
     lock  sync.Mutex
     cond  *sync.Cond
 }
 ?
 func NewQueue(size int) *Queue {
     q := &Queue{
         items: make([]int, 0, size),
         size:  size,
     }
     q.cond = sync.NewCond(&q.lock)
     return q
 }
 ?
 func (q *Queue) Put(item int) {
     q.lock.Lock()
     defer q.lock.Unlock()
 ?
     for len(q.items) == q.size {
         q.cond.Wait()
     }
 ?
     q.items = append(q.items, item)
     fmt.Printf("put item %d, queue len %d\n", item, len(q.items))
 ?
     q.cond.Signal()
 }
 ?
 func (q *Queue) Get() int {
     q.lock.Lock()
     defer q.lock.Unlock()
 ?
     for len(q.items) == 0 {
         q.cond.Wait()
     }
 ?
     item := q.items[0]
     q.items = q.items[1:]
     fmt.Printf("get item %d, queue len %d\n", item, len(q.items))
 ?
     q.cond.Signal()
     return item
 }
 ?
 func Producer(q *Queue, id int) {
     for {
         item := rand.Intn(100)
         q.Put(item)
         time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
     }
 }
 ?
 func Consumer(q *Queue, id int) {
     for {
         item := q.Get()
         time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
     }
 }
 ?
 func main() {
     q := NewQueue(5)
     for i := 0; i < 3; i++ {
         go Producer(q, i)
     }
     for i := 0; i < 5; i++ {
         go Consumer(q, i)
     }
     time.Sleep(10 * time.Second)
 }

在這個(gè)例子中，我們創(chuàng)建了一個(gè) Queue 類型，它包含一個(gè)整數(shù)數(shù)組和一個(gè)長(zhǎng)度。在 Put 和 Get 方法中，我們使用互斥鎖和條件變量進(jìn)行同步。

在 Producer 和 Consumer 函數(shù)中，我們模擬生產(chǎn)者和消費(fèi)者的行為。生產(chǎn)者會(huì)不斷地生成隨機(jī)數(shù)，并調(diào)用 Put 方法將其放入隊(duì)列中；消費(fèi)者會(huì)不斷地調(diào)用 Get 方法從隊(duì)列中取出數(shù)據(jù)。

在主函數(shù)中，我們創(chuàng)建了多個(gè)生產(chǎn)者和消費(fèi)者 goroutine，它們并發(fā)地操作隊(duì)列。在程序運(yùn)行過程中，我們可以看到隊(duì)列的長(zhǎng)度會(huì)不斷地變化，生產(chǎn)者和消費(fèi)者會(huì)交替執(zhí)行。

5. 總結(jié)

sync.Cond 是 Go 語(yǔ)言中非常重要的同步原語(yǔ)之一。它可以幫助我們實(shí)現(xiàn)更高級(jí)別的同步機(jī)制，例如生產(chǎn)者和消費(fèi)者模型、讀寫鎖等。同時(shí)，它也是一個(gè)非常復(fù)雜的數(shù)據(jù)結(jié)構(gòu)，需要深入理解其內(nèi)部實(shí)現(xiàn)才能正確地使用它。

在使用 sync.Cond 時(shí)，我們需要注意以下幾點(diǎn)：

在使用 sync.Cond 前，一定要先創(chuàng)建一個(gè)互斥鎖。
在調(diào)用 Wait 方法前，一定要先獲取互斥鎖，否則會(huì)導(dǎo)致死鎖。
在調(diào)用 Wait 方法后，當(dāng)前 goroutine 會(huì)被阻塞，直到被喚醒。
在調(diào)用 Signal 或 Broadcast 方法后，等待隊(duì)列中的一個(gè)或多個(gè) goroutine 會(huì)被喚醒，但不會(huì)立即獲取互斥鎖。因此，在使用 Signal 或 Broadcast 方法時(shí)，一定要保證喚醒的 goroutine 不會(huì)互相競(jìng)爭(zhēng)同一個(gè)資源。
在調(diào)用 Signal 或 Broadcast 方法后，一定要釋放互斥鎖，否則被喚醒的 goroutine 無(wú)法獲取到互斥鎖，仍然會(huì)被阻塞。
在使用 sync.Cond 時(shí)，一定要注意競(jìng)爭(zhēng)條件和數(shù)據(jù)同步的問題，確保程序的正確性和穩(wěn)定性。

在本文中，我們介紹了 sync.Cond 的基本用法和內(nèi)部實(shí)現(xiàn)原理，并通過一個(gè)實(shí)際的生產(chǎn)者和消費(fèi)者模型的例子，展示了如何使用 sync.Cond 實(shí)現(xiàn)高級(jí)別的同步機(jī)制。

使用 sync.Cond 可以幫助我們實(shí)現(xiàn)更高效、更靈活、更安全的并發(fā)程序。但同時(shí)，也需要我們仔細(xì)思考和理解其內(nèi)部實(shí)現(xiàn)，避免出現(xiàn)競(jìng)爭(zhēng)條件和數(shù)據(jù)同步的問題，確保程序的正確性和穩(wěn)定性。

總之，Golang 的 sync.Cond 類型是 Golang 并發(fā)編程中非常重要的一個(gè)組件，熟練掌握它的使用方法和實(shí)現(xiàn)原理，可以有效提升 Golang 并發(fā)編程的能力和水平。

以上就是深入學(xué)習(xí)Golang并發(fā)編程必備利器之sync.Cond類型的詳細(xì)內(nèi)容，更多關(guān)于Golang sync.Cond的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: