Go并发编程sync.Cond的具体使用

第Go并发编程sync.Cond的具体使用目录简介详细介绍案例：Redis连接池注意点

简介

Go标准库提供Cond原语的目的是，为等待/通知场景下的并发问题提供支持。Cond通常应用于等待某个条件的一组goroutine，等条件变为true的时候，其中一个goroutine或者所有的goroutine都会被唤醒执行。

Cond是和某个条件相关，这个条件需要一组goroutine协作共同完成，在条件还没有满足的时候，所有等待这个条件的goroutine都会被阻塞住，只有这一组goroutine通过协作达到了这个条件，等待的goroutine才可能继续进行下去。

这个条件可以是我们自定义的true/false逻辑表达式。

但是Cond使用的比较少，因为在大部分场景下是可以被Channel和WaitGroup来替换的。

详细介绍

下面就是Cond的数据结构和对外提供的方法，Cond内部维护了一个等待队列和锁实例。

typeCondstruct{

noCopynoCopy

//锁

LLocker

//等待队列

notifynotifyList

checkercopyChecker

funcNeWCond(lLocker)*Cond

func(c*Cond)Broadcast()

func(c*Cond)Signal()

func(c*Cond)Wait()

NeWCond：NeWCond方法需要调用者传入一个Locker接口，这个接口就Lock/UnLock方法，所以我们可以传入一个sync.Metex对象

Signal：允许调用者唤醒一个等待当前Cond的goroutine。如果Cond等待队列中有一个或者多个等待的goroutine，则从等待队列中移除第一个goroutine并把它唤醒

Broadcast：允许调用者唤醒所有等待当前Cond的goroutine。如果Cond等待队列中有一个或者多个等待的goroutine，则清空所有等待的goroutine，并全部唤醒

Wait：会把调用者放入Cond的等待队列中并阻塞，直到被Signal或者Broadcast的方法从等待队列中移除并唤醒

案例：Redis连接池

可以看一下下面的代码，使用了Cond实现一个Redis的连接池，最关键的代码就是在链表为空的时候需要调用Cond的Wait方法，将gorutine进行阻塞。然后goruntine在使用完连接后，将连接返回池子后，需要通知其他阻塞的goruntine来获取连接。

packagemain

import(

"container/list"

"fmt"

"math/rand"

"sync"

"time"

//连接池

typePoolstruct{

locksync.Mutex//锁

clientslist.List//连接

cond*sync.Cond//cond实例

closebool//是否关闭

//RedisClient

typeClientstruct{

idint32

//创建RedisClient

funcNewClient()*Client{

returnClient{

id:rand.Int31n(100000),

//关闭RedisClient

func(this*Client)Close(){

fmt.Printf("Client:%d正在关闭",this.id)

//创建连接池

funcNewPool(maxConnNumint)*Pool{

pool:=new(Pool)

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

//创建连接

fori:=0;imaxConnNum;i++{

client:=NewClient()

pool.clients.PushBack(client)

returnpool

//从池子中获取连接

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

this.lock.Lock()

deferthis.lock.Unlock()

//已关闭

ifthis.close{

fmt.Println("Poolisclosed")

returnnil

//如果连接池没有连接需要阻塞

forthis.clients.Len()=0{

this.cond.Wait()

//从链表中取出头节点，删除并返回

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

//向链表尾部插入一个连接

this.clients.PushBack(client)

//唤醒一个正在等待的goruntine

this.cond.Signal()

//关闭池子

func(this*Pool)Close(){

this.lock.Lock()

deferthis.lock.Unlock()

//关闭连接

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

client:=e.Value.(*Client)

client.Close()

//重置数据

this.close=true

this.clients.Init()

funcmain(){

varwgsync.WaitGroup

pool:=NewPool(3)

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

wg.Add(1)

gofunc(indexint){

deferwg.Done()

//获取一个连接

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()

}

运行结果：

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]放回池子

注意点

在调用Wait方法前，需要先加锁，就像我上面例子中Pull方法也是先加锁

看一下源码就知道了，因为Wait方法的执行逻辑是先将goruntine添加到等待队列中，然后释放锁，然后阻塞，等唤醒后，会继续加锁。如果在调用Wait前不加锁，但是里面会解锁，执行的时候就会报错。

//

//c.L.Lock()

//for!condition(){

//c.Wait()

//...makeuseofcondition...

//c.L.Unlock()

func(c*Cond)Wait(){

c.checker.check()

//添加到等待队列

t:=runtime_notifyListAdd(c.notify)

c.L.Unlock()

//阻塞

runtime_notifyListWait(c.notify,t)

c.L.Lock()

}

还是Wait方法，在唤醒后需要继续检查Cond条件

就拿上面的redis连接案例来进行说明吧，我这里是使用了for循环来进行检测。如果将for循环改成使用if，也就是只判断一次，会有什么问题？可以停下来先想想

上面说了调用者也可以使用Broadcast方法来唤醒goruntine，如果使用