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

type cond struct {
   nocopy nocopy

   // 锁
   l locker

   // 等待队列
   notify  notifylist
   checker copychecker
}

func newcond(l locker) *cond
func (c *cond) broadcast()
func (c *cond) signal()
func (c *cond) wait()

package main

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

// 连接池
type pool struct {
   lock    sync.mutex // 锁
   clients list.list  // 连接
   cond    *sync.cond // cond实例
   close   bool       // 是否关闭
}

// redis client
type client struct {
   id int32
}

// 创建redis client
func newclient() *client {
   return &client{
      id: rand.int31n(100000),
   }
}

// 关闭redis client
func (this *client) close() {
   fmt.printf("client:%d 正在关闭", this.id)
}

// 创建连接池
func newpool(maxconnnum int) *pool {
   pool := new(pool)
   pool.cond = sync.newcond(&pool.lock)

   // 创建连接
   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()

   // 已关闭
   if this.close {
      fmt.println("pool is closed")
      return nil
   }

   // 如果连接池没有连接 需要阻塞
   for this.clients.len() <= 0 {
      this.cond.wait()
   }

   // 从链表中取出头节点，删除并返回
   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
   }

   // 向链表尾部插入一个连接
   this.clients.pushback(client)

   // 唤醒一个正在等待的goruntine
   this.cond.signal()
}

// 关闭池子
func (this *pool) close() {
   this.lock.lock()
   defer this.lock.unlock()

   // 关闭连接
   for e := this.clients.front(); e != nil; e = e.next() {
      client := e.value.(*client)
      client.close()
   }

   // 重置数据
   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()

         // 获取一个连接
         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()
}

//
//    c.l.lock()
//    for !condition() {
//        c.wait()
//    }
//    ... make use of condition ...
//    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()
}

// 如果连接池没有连接 需要阻塞
for this.clients.len() <= 0 {
  this.cond.wait()
}

// 获取连接
ele := this.clients.remove(this.clients.front())
return ele.(*client)