使用Go语言实现一个简单的无界资源池

// Pool 定义一个结构体 包含重量级锁 有缓冲区Chanel 工厂方法 连接池关闭状态
type Pool struct {
 m        sync.Mutex
 resource chan io.Closer
 factory  func() (io.Closer, error)
 closed   bool
}

var ErrPoolClosed = errors.New("连接池已关闭")


func New(fn func() (io.Closer, error), size uint) (*Pool, error) {
 //判断size大小是否合法
 if size <= 0 {
  return nil, errors.New("size不合法")
 }
 //基于工厂方法和size创建资源池
 return &Pool{
  resource: make(chan io.Closer, size),
  factory:  fn,
 }, nil

}

func (p *Pool) Acquire() (io.Closer, error) {
 select {
 //如果channel有足够的资源分配则直接返回
 case r, ok := <-p.resource:
  if !ok {
   log.Println("连接池已关闭")
   return nil, ErrPoolClosed
  }
  log.Println("拿到连接池共享资源")
  return r, nil
  //基于工厂方法创建全新的资源返回出去
 default:
  log.Println("资源不足，创建新的连接资源")
  return p.factory()
 }

}

// Release 上锁 设置方法退出后解锁 查看当前连接池是否已关闭，若关闭则直接将资源关闭 ，反之select查看能否将其存入缓冲区，若可以输出入队成功，反之输出队列已满
func (p *Pool) Release(r io.Closer) {
 //上锁确保关闭和归还资源操作互斥
 p.m.Lock()
 //函数退出时解锁
 defer p.m.Unlock()
 //如果资源池关闭则直接将当前资源关闭销毁
 if p.closed {
  log.Println("连接池已关闭,直接销毁当前资源")
  r.Close()
 }
 //将连接归还，如果满了则直接关闭销毁
 select {
 case p.resource <- r:
  log.Println("连接归还成功")
 default:
  log.Println("连接池已满，资源直接销毁")
  r.Close()
 }

}

// Close 方法 上锁 设置方法退出后解锁 遍历所有资源将其关闭 然后再关闭连接池
func (p *Pool) Close() {
 p.m.Lock()

 defer p.m.Unlock()

 if p.closed {
  log.Println("连接池已关闭,直接销毁当前资源")
  return
 }
 //设置为关闭
 p.closed = true
 //关闭资源
 close(p.resource)
 //遍历资源池资源
 for r := range p.resource {
  r.Close()
 }

}

//设置最大协程数与资源池数为24
const maxGoroutines = 24
const poolResources = 24

//创建可关闭的数据库连接
type dbConnection struct {
 ID int32
}
//对应的关闭方法
func (d *dbConnection) Close() error {
 log.Println("当前数据库连接", d.ID, "已关闭")
 return nil
}

var idCounter int32

func createConnection() (io.Closer, error) {
 id := atomic.AddInt32(&idCounter, 1)
 return &dbConnection{ID: id}, nil
}

func main() {
 //创建maxGoroutines个WaitGroup
 var wg sync.WaitGroup
 wg.Add(maxGoroutines)
 //传入createConnection方法和连接池大小poolResources创建数据库连接池
 p, err := pool.New(createConnection, poolResources)
 if err != nil {
  log.Println(err)
 }
 //创建24个协程获取资源
 for i := 0; i < maxGoroutines; i++ {
  go func(queryParam int) {
   queryData(queryParam, p)
   defer wg.Done()
  }(i)
 }
 //等待操作完成关闭连接池
 wg.Wait()
 log.Println("查询完成")
 p.Close()

}
//queryData 基于连接池Acquire获取资源，完成后通过Release归还资源
func queryData(queryParam int, p *pool.Pool) {

 r, e := p.Acquire()
 if e != nil {
  log.Println(e)
  return
 }

 defer p.Release(r)

 time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
 log.Println("查询", queryParam, "使用连接", r.(*dbConnection).ID)
}