第Go并發(fā)編程sync.Cond的具體使用目錄簡(jiǎn)介詳細(xì)介紹案例：Redis連接池注意點(diǎn)

簡(jiǎn)介

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

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

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

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

詳細(xì)介紹

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

typeCondstruct{

noCopynoCopy

//鎖

LLocker

//等待隊(duì)列

notifynotifyList

checkercopyChecker

funcNeWCond(lLocker)*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來獲取連接。

packagemain

import(

"container/list"

"fmt"

"math/rand"

"sync"

"time"

//連接池

typePoolstruct{

locksync.Mutex//鎖

clientslist.List//連接

cond*sync.Cond//cond實(shí)例

closebool//是否關(guān)閉

//RedisClient

typeClientstruct{

idint32

//創(chuàng)建RedisClient

funcNewClient()*Client{

returnClient{

id:rand.Int31n(100000),

//關(guān)閉RedisClient

func(this*Client)Close(){

fmt.Printf("Client:%d正在關(guān)閉",this.id)

//創(chuàng)建連接池

funcNewPool(maxConnNumint)*Pool{

pool:=new(Pool)

pool.cond=sync.NewCond(pool.lock)

//創(chuàng)建連接

fori:=0;imaxConnNum;i++{

client:=NewClient()

pool.clients.PushBack(client)

returnpool

//從池子中獲取連接

func(this*Pool)Pull()*Client{

this.lock.Lock()

deferthis.lock.Unlock()

//已關(guān)閉

ifthis.close{

fmt.Println("Poolisclosed")

returnnil

//如果連接池沒有連接需要阻塞

forthis.clients.Len()=0{

this.cond.Wait()

//從鏈表中取出頭節(jié)點(diǎn)，刪除并返回

ele:=this.clients.Remove(this.clients.Front())

returnele.(*Client)

//將連接放回池子

func(this*Pool)Push(client*Client){

this.lock.Lock()

deferthis.lock.Unlock()

ifthis.close{

fmt.Println("Poolisclosed")

return

//向鏈表尾部插入一個(gè)連接

this.clients.PushBack(client)

//喚醒一個(gè)正在等待的goruntine

this.cond.Signal()

//關(guān)閉池子

func(this*Pool)Close(){

this.lock.Lock()

deferthis.lock.Unlock()

//關(guān)閉連接

fore:=this.clients.Front();e!=nil;e=e.Next(){

client:=e.Value.(*Client)

client.Close()

//重置數(shù)據(jù)

this.close=true

this.clients.Init()

funcmain(){

varwgsync.WaitGroup

pool:=NewPool(3)

fori:=1;i=10;i++{

wg.Add(1)

gofunc(indexint){

deferwg.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)閃ait方法的執(zhí)行邏輯是先將goruntine添加到等待隊(duì)列中，然后釋放鎖，然后阻塞，等喚醒后，會(huì)繼續(xù)加鎖。如果在調(diào)用Wait前不加鎖，但是里面會(huì)解鎖，執(zhí)行的時(shí)候就會(huì)報(bào)錯(cuò)。

//

//c.L.Lock()

//for!condition(){

//c.Wait()

//...makeuseofcondition...

//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連接案例來進(jìn)行說明吧，我這里是使用了for循環(huán)來進(jìn)行檢測(cè)。如果將for循環(huán)改成使用if，也就是只判斷一次，會(huì)有什么問題？可以停下來先想想

上面說了調(diào)用者也可以使用Broadcast方法來喚醒goruntine，如果使用