簡(jiǎn)介

Go 標(biāo)準(zhǔn)庫(kù)提供 Cond 原語(yǔ)的目的是，為等待 / 通知場(chǎng)景下的并發(fā)問(wèn)題提供支持。Cond 通常應(yīng)用于等待某個(gè)條件的一組 goroutine，等條件變?yōu)?true 的時(shí)候，其中一個(gè) goroutine 或者所有的 goroutine 都會(huì)被喚醒執(zhí)行。

Cond 是和某個(gè)條件相關(guān)，這個(gè)條件需要一組 goroutine 協(xié)作共同完成，在條件還沒(méi)有滿(mǎn)足的時(shí)候，所有等待這個(gè)條件的 goroutine 都會(huì)被阻塞住，只有這一組 goroutine 通過(guò)協(xié)作達(dá)到了這個(gè)條件，等待的 goroutine 才可能繼續(xù)進(jìn)行下去。

這個(gè)條件可以是我們自定義的 true/false 邏輯表達(dá)式。

但是 Cond 使用的比較少，因?yàn)樵诖蟛糠謭?chǎng)景下是可以被 Channel 和 WaitGroup 來(lái)替換的。

詳細(xì)介紹

下面就是 Cond 的數(shù)據(jù)結(jié)構(gòu)和對(duì)外提供的方法，Cond 內(nèi)部維護(hù)了一個(gè)等待隊(duì)列和鎖實(shí)例。

type Cond struct {
   noCopy noCopy

   // 鎖
   L Locker

   // 等待隊(duì)列
   notify  notifyList
   checker copyChecker
}

func NeWCond(l Locker) *Cond
func (c *Cond) Broadcast()
func (c *Cond) Signal()
func (c *Cond) Wait()

• NeWCond： NeWCond 方法需要調(diào)用者傳入一個(gè) Locker 接口，這個(gè)接口就 Lock/UnLock 方法，所以我們可以傳入一個(gè) sync.Metex 對(duì)象

• Signal：允許調(diào)用者喚醒一個(gè)等待當(dāng)前 Cond 的 goroutine。如果 Cond 等待隊(duì)列中有一個(gè)或者多個(gè)等待的 goroutine ，則從等待隊(duì)列中移除第一個(gè) goroutine 并把它喚醒

• Broadcast：允許調(diào)用者喚醒所有等待當(dāng)前 Cond 的 goroutine。如果 Cond 等待隊(duì)列中有一個(gè)或者多個(gè)等待的 goroutine，則清空所有等待的 goroutine，并全部喚醒

• Wait：會(huì)把調(diào)用者放入 Cond 的等待隊(duì)列中并阻塞，直到被 Signal 或者 Broadcast 的方法從等待隊(duì)列中移除并喚醒

案例：Redis連接池

可以看一下下面的代碼，使用了 Cond 實(shí)現(xiàn)一個(gè) Redis 的連接池，最關(guān)鍵的代碼就是在鏈表為空的時(shí)候需要調(diào)用 Cond 的 Wait 方法，將 gorutine 進(jìn)行阻塞。然后 goruntine 在使用完連接后，將連接返回池子后，需要通知其他阻塞的 goruntine 來(lái)獲取連接。

package main

import (
   "container/list"
   "fmt"
   "math/rand"
   "sync"
   "time"
)

// 連接池
type Pool struct {
   lock    sync.Mutex // 鎖
   clients list.List  // 連接
   cond    *sync.Cond // cond實(shí)例
   close   bool       // 是否關(guān)閉
}

// Redis Client
type Client struct {
   id int32
}

// 創(chuàng)建Redis Client
func NewClient() *Client {
   return &Client{
      id: rand.Int31n(100000),
   }
}

// 關(guān)閉Redis Client
func (this *Client) Close() {
   fmt.Printf("Client:%d 正在關(guān)閉", this.id)
}

// 創(chuàng)建連接池
func NewPool(maxConnNum int) *Pool {
   pool := new(Pool)
   pool.cond = sync.NewCond(&pool.lock)

   // 創(chuàng)建連接
   for i := 0; i < maxConnNum; i++ {
      client := NewClient()
      pool.clients.PushBack(client)
   }

   return pool
}

// 從池子中獲取連接
func (this *Pool) Pull() *Client {
   this.lock.Lock()
   defer this.lock.Unlock()

   // 已關(guān)閉
   if this.close {
      fmt.Println("Pool is closed")
      return nil
   }

   // 如果連接池沒(méi)有連接 需要阻塞
   for this.clients.Len() <= 0 {
      this.cond.Wait()
   }

   // 從鏈表中取出頭節(jié)點(diǎn)，刪除并返回
   ele := this.clients.Remove(this.clients.Front())
   return ele.(*Client)
}

// 將連接放回池子
func (this *Pool) Push(client *Client) {
   this.lock.Lock()
   defer this.lock.Unlock()

   if this.close {
      fmt.Println("Pool is closed")
      return
   }

   // 向鏈表尾部插入一個(gè)連接
   this.clients.PushBack(client)

   // 喚醒一個(gè)正在等待的goruntine
   this.cond.Signal()
}

// 關(guān)閉池子
func (this *Pool) Close() {
   this.lock.Lock()
   defer this.lock.Unlock()

   // 關(guān)閉連接
   for e := this.clients.Front(); e != nil; e = e.Next() {
      client := e.Value.(*Client)
      client.Close()
   }

   // 重置數(shù)據(jù)
   this.close = true
   this.clients.Init()
}

func main() {

   var wg sync.WaitGroup

   pool := NewPool(3)
   for i := 1; i <= 10; i++ {
      wg.Add(1)
      go func(index int) {

         defer wg.Done()

         // 獲取一個(gè)連接
         client := pool.Pull()

         fmt.Printf("Time:%s | 【goruntine#%d】獲取到client[%d]\n", time.Now().Format("15:04:05"), index, client.id)
         time.Sleep(time.Second * 5)
         fmt.Printf("Time:%s | 【goruntine#%d】使用完畢，將client[%d]放回池子\n", time.Now().Format("15:04:05"), index, client.id)

         // 將連接放回池子
         pool.Push(client)
      }(i)
   }

   wg.Wait()
}

運(yùn)行結(jié)果：

Time:15:10:25 | 【goruntine#7】獲取到client[31847]
Time:15:10:25 | 【goruntine#5】獲取到client[27887]
Time:15:10:25 | 【goruntine#10】獲取到client[98081]
Time:15:10:30 | 【goruntine#5】使用完畢，將client[27887]放回池子
Time:15:10:30 | 【goruntine#6】獲取到client[27887]               
Time:15:10:30 | 【goruntine#10】使用完畢，將client[98081]放回池子
Time:15:10:30 | 【goruntine#7】使用完畢，將client[31847]放回池子 
Time:15:10:30 | 【goruntine#1】獲取到client[31847]               
Time:15:10:30 | 【goruntine#9】獲取到client[98081]               
Time:15:10:35 | 【goruntine#6】使用完畢，將client[27887]放回池子
Time:15:10:35 | 【goruntine#3】獲取到client[27887]              
Time:15:10:35 | 【goruntine#1】使用完畢，將client[31847]放回池子
Time:15:10:35 | 【goruntine#4】獲取到client[31847]              
Time:15:10:35 | 【goruntine#9】使用完畢，將client[98081]放回池子
Time:15:10:35 | 【goruntine#2】獲取到client[98081]              
Time:15:10:40 | 【goruntine#3】使用完畢，將client[27887]放回池子
Time:15:10:40 | 【goruntine#8】獲取到client[27887]              
Time:15:10:40 | 【goruntine#2】使用完畢，將client[98081]放回池子
Time:15:10:40 | 【goruntine#4】使用完畢，將client[31847]放回池子
Time:15:10:45 | 【goruntine#8】使用完畢，將client[27887]放回池子

注意點(diǎn)

• 在調(diào)用 Wait 方法前，需要先加鎖，就像我上面例子中 Pull 方法也是先加鎖

看一下源碼就知道了，因?yàn)?Wait 方法的執(zhí)行邏輯是先將 goruntine 添加到等待隊(duì)列中，然后釋放鎖，然后阻塞，等喚醒后，會(huì)繼續(xù)加鎖。如果在調(diào)用 Wait 前不加鎖，但是里面會(huì)解鎖，執(zhí)行的時(shí)候就會(huì)報(bào)錯(cuò)。

//
//    c.L.Lock()
//    for !condition() {
//        c.Wait()
//    }
//    ... make use of condition ...
//    c.L.Unlock()
//
func (c *Cond) Wait() {
   c.checker.check()
   
   // 添加到等待隊(duì)列
   t := runtime_notifyListAdd(&c.notify)
   c.L.Unlock()
   
   // 阻塞
   runtime_notifyListWait(&c.notify, t)
   c.L.Lock()
}

• 還是 Wait 方法，在喚醒后需要繼續(xù)檢查 Cond 條件

就拿上面的 redis 連接案例來(lái)進(jìn)行說(shuō)明吧，我這里是使用了 for 循環(huán)來(lái)進(jìn)行檢測(cè)。如果將 for 循環(huán)改成使用 if，也就是只判斷一次，會(huì)有什么問(wèn)題？可以停下來(lái)先想想

上面說(shuō)了調(diào)用者也可以使用 Broadcast 方法來(lái)喚醒 goruntine ，如果使用的是 Broadcast 方法，所有的 goruntine 都會(huì)被喚醒，然后大家都去鏈表中去獲取 redis 連接了，就會(huì)出現(xiàn)部分 goruntine拿不到連接，實(shí)際上沒(méi)有那么多連接可以獲取，因?yàn)槊看沃粫?huì)放回一個(gè)連接到池子中。

// 如果連接池沒(méi)有連接 需要阻塞
for this.clients.Len() <= 0 {
  this.cond.Wait()
}

// 獲取連接
ele := this.clients.Remove(this.clients.Front())
return ele.(*Client)

到此這篇關(guān)于Go并發(fā)編程sync.Cond的具體使用的文章就介紹到這了,更多相關(guān)Go sync.Cond內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論