当前位置 博文首页 > Golang连接池的几种实现案例小结
因为TCP的三只握手等等原因，建立一个连接是一件成本比较高的行为。所以在一个需要多次与特定实体交互的程序中，就需要维持一个连接池，里面有可以复用的连接可供重复使用。
而维持一个连接池，最基本的要求就是要做到：thread safe（线程安全），尤其是在Golang这种特性是goroutine的语言中。
type Pool struct {
m sync.Mutex // 保证多个goroutine访问时候，closed的线程安全
res chan io.Closer //连接存储的chan
factory func() (io.Closer,error) //新建连接的工厂方法
closed bool //连接池关闭标志
}
这个简单的连接池，我们利用chan来存储池里的连接。而新建结构体的方法也比较简单：
func New(fn func() (io.Closer, error), size uint) (*Pool, error) {
if size <= 0 {
return nil, errors.New("size的值太小了。")
}
return &Pool{
factory: fn,
res: make(chan io.Closer, size),
}, nil
}
只需要提供对应的工厂函数和连接池的大小就可以了。
获取连接
那么我们要怎么从中获取资源呢？因为我们内部存储连接的结构是chan，所以只需要简单的select就可以保证线程安全：
//从资源池里获取一个资源
func (p *Pool) Acquire() (io.Closer,error) {
select {
case r,ok := <-p.res:
log.Println("Acquire:共享资源")
if !ok {
return nil,ErrPoolClosed
}
return r,nil
default:
log.Println("Acquire:新生成资源")
return p.factory()
}
}
我们先从连接池的res这个chan里面获取，如果没有的话我们就利用我们早已经准备好的工厂函数进行构造连接。同时我们在从res获取连接的时候利用ok先确定了这个连接池是否已经关闭。如果已经关闭的话我们就返回早已经准备好的连接已关闭错误。
关闭连接池
那么既然提到关闭连接池，我们是怎么样关闭连接池的呢？
//关闭资源池，释放资源
func (p *Pool) Close() {
p.m.Lock()
defer p.m.Unlock()
if p.closed {
return
}
p.closed = true
//关闭通道，不让写入了
close(p.res)
//关闭通道里的资源
for r:=range p.res {
r.Close()
}
}
这边我们需要先进行p.m.Lock()上锁操作，这么做是因为我们需要对结构体里面的closed进行读写。需要先把这个标志位设定后，关闭res这个chan，使得Acquire方法无法再获取新的连接。我们再对res这个chan里面的连接进行Close操作。
释放连接
释放连接首先得有个前提，就是连接池还没有关闭。如果连接池已经关闭再往res里面送连接的话就好触发panic。
func (p *Pool) Release(r io.Closer){
//保证该操作和Close方法的操作是安全的
p.m.Lock()
defer p.m.Unlock()
//资源池都关闭了，就省这一个没有释放的资源了，释放即可
if p.closed {
r.Close()
return
}
select {
case p.res <- r:
log.Println("资源释放到池子里了")
default:
log.Println("资源池满了，释放这个资源吧")
r.Close()
}
}
现在基本就解决前面说的小问题了。不会出现连接太多导致无法控制too many connections的情况。也很好了维持了连接池的最小数量。同时也做了相关对于连接健康性的检查操作。
值得注意的是，作为标准库的代码，相关注释和代码都非常完美，真的可以看的神清气爽。
这个Golang实现的Redis客户端，是怎么实现连接池的。这边的思路非常奇妙，还是能学习到不少好思路。当然了，由于代码注释比较少，啃起来第一下还是有点迷糊的。相关代码地址在https://github.com/go-redis/redis/blob/master/internal/pool/pool.go 可以看到。
而它的连接池结构如下
type ConnPool struct {
...
queue chan struct{}
connsMu sync.Mutex
conns []*Conn
idleConns []*Conn
poolSize int
idleConnsLen int
stats Stats
_closed uint32 // atomic
closedCh chan struct{}
}
我们可以看到里面存储连接的结构还是slice。但是我们可以重点看看queue，conns，idleConns这几个变量，后面会提及到。但是值得注意的是！我们可以看到，这里有两个**[]Conn**结构：conns、idleConns，那么问题来了：
到底连接存在哪里？
新建连接池连接
我们先从新建连接池连接开始看：
func NewConnPool(opt *Options) *ConnPool {
....
p.checkMinIdleConns()
if opt.IdleTimeout > 0 && opt.IdleCheckFrequency > 0 {
go p.reaper(opt.IdleCheckFrequency)
}
....
}
初始化连接池的函数有个和前面两个不同的地方。
获取连接
func (p *ConnPool) Get(ctx context.Context) (*Conn, error) {
if p.closed() {
return nil, ErrClosed
}
//这边和前面sql获取连接函数的流程先不同。sql是先看看连接池有没有空闲连接，有的话先获取不到再排队。这边是直接先排队获取令牌，排队函数后面会分析。
err := p.waitTurn(ctx)
if err != nil {
return nil, err
}
//前面没出error的话，就已经排队轮候到了。接下来就是获取的流程。
for {
p.connsMu.Lock()
//从空闲连接里面先获取一个空闲连接。
cn := p.popIdle()
p.connsMu.Unlock()
if cn == nil {
// 没有空闲连接时候直接跳出循环。
break
}
// 判断是否已经过时，是的话close掉了然后继续取出。
if p.isStaleConn(cn) {
_ = p.CloseConn(cn)
continue
}
atomic.AddUint32(&p.stats.Hits, 1)
return cn, nil
}
atomic.AddUint32(&p.stats.Misses, 1)
// 如果没有空闲连接的话，这边就直接新建连接了。
newcn, err := p.newConn(ctx, true)
if err != nil {
// 归还令牌。
p.freeTurn()
return nil, err
}
return newcn, nil
}
我们可以试着回答开头那个问题：连接到底存在哪里？答案是从cn := p.popIdle()这句话可以看出，获取连接这个动作，是从idleConns里面获取的，而里面的函数也证明了这一点。但是，真的是这样的嘛？我们后面再看看。
同时我的理解是：
而：
func (p *ConnPool) freeTurn() {
<-p.queue
}
func (p *ConnPool) waitTurn(ctx context.Context) error {
...
case p.queue <- struct{}{}:
return nil
...
}
就是在靠queue这个chan来维持令牌数量。
那么conns的作用是什么呢？我们可以来看看新建连接这个函数：
新建连接
func (p *ConnPool) newConn(ctx context.Context, pooled bool) (*Conn, error) {
cn, err := p.dialConn(ctx, pooled)
if err != nil {
return nil, err
}
p.connsMu.Lock()
p.conns = append(p.conns, cn)
if pooled {
// 如果连接池满了，会在后面移除。
if p.poolSize >= p.opt.PoolSize {
cn.pooled = false
} else {
p.poolSize++
}
}
p.connsMu.Unlock()
return cn, nil
}
基本逻辑出来了。就是如果新建连接的话，我并不会直接放在idleConns里面，而是先放conns里面。同时先看池子满了没有。满的话后面归还的时候会标记，后面会删除。那么这个后面会删除，指的是什么时候呢？那就是下面说的归还连接的时候了。
归还连接
func (p *ConnPool) Put(cn *Conn) {
if cn.rd.Buffered() > 0 {
internal.Logger.Printf("Conn has unread data")
p.Remove(cn, BadConnError{})
return
}
//这就是我们刚刚说的后面了，前面标记过不要入池的，这边就删除了。当然了，里面也会进行freeTurn操作。
if !cn.pooled {
// 这个方法就是前面的标志位，判断里面可以知道，前面标志不要池化的，这里会将它删除。
p.Remove(cn, nil)
return
}
p.connsMu.Lock()
p.idleConns = append(p.idleConns, cn)
p.idleConnsLen++
p.connsMu.Unlock()
//我们可以看到很明显的这个归还号码牌的动作。
p.freeTurn()
}
答案就是，所有的连接其实是存放在conns这个切片里面。如果这个连接是空闲等待的状态的话，那就在idleConns里面加一个自己的指针！
其实归还的过程，就是检查一下我打算还的这个连接，是不是超售的产物，如果是就没必要池化了，直接删除就可以了。不是的话，就是把连接自身（一个指针）在idleConns也append一下。
等等，上面的逻辑似乎有点不对？我们来理一下获取连接流程：
我当时疑惑了好久，既然始终都需要获得令牌才能得到连接，令牌数量是定的。为什么还会超卖呢？翻了一下源码，我的答案是：
虽然Get方法获取连接是newConn这个私用方法，受到令牌管制导致不会出现超卖。但是这个方法接受传参：pooled bool。所以我猜是担心其他人调用这个方法时候，不管三七二十一就传了true，导致poolSize越来越大。
总的来说，redis这个连接池的连接数控制，还是在queue这个我称为令牌的chan进行操作。
上面可以看到，连接池的最基本的保证，就是获取连接时候的线程安全。但是在实现诸多额外特性时候却又从不同角度来实现。还是非常有意思的。但是不管存储结构是用chan还是还是slice，都可以很好的实现这一点。如果像sql或者redis那样用slice来存储连接，就得维护一个结构来表示排队等候的效果。
js