Redigo 连接池的使用
大家都知道go语言中的goroutine虽然消耗资源很小，并且是一个用户线程。但是goroutine也不是无限开的，所以我们会有很多关于协程池的库，当然啊我们自己也可以完成一些简单的携程池。redis也是相同的，redis的链接也是不推荐无限制的打开，否则会造成redis负荷加重。
先看一下Redigo 中的连接池的使用
package mainimport ( "fmt" "github.com/panlei/redigo/redis" "time")func main() { pool := &redis.Pool{ MaxIdle: 4, MaxActive: 4, Dial: func() (redis.Conn, error) { rc, err := redis.Dial("tcp", "127.0.0.1:6379") if err != nil { return nil, err } return rc, nil }, IdleTimeout: time.Second, Wait: true, } con := pool.Get() str, err := redis.String(con.Do("get", "aaa")) con.Close() fmt.Println("value: ", str, " err:", err)}我们可以看到Redigo使用连接池还是很简单的步骤：
- 创建连接池
- 简单设置连接池的最大链接数等参数
- 注入拨号函数（设置redis地址 端口号等）
- 调用pool.Get() 获取连接
- 使用连接Do函数请求redis
- 关闭连接
源码
Pool conn 对象的定义
type Pool struct { // 拨号函数 从外部注入 Dial func() (Conn, error) // DialContext is an application supplied function for creating and configuring a DialContext func(ctx context.Context) (Conn, error) // 检测连接的可用性，从外部注入。如果返回error 则直接关闭连接 TestOnBorrow func(c Conn, t time.Time) error // 最大闲置连接数量 MaxIdle int // 最大活动连接数 MaxActive int // 闲置过期时间 在get函数中会有逻辑 删除过期的连接 IdleTimeout time.Duration // 设置如果活动连接达到上限 再获取时候是等待还是返回错误 // 如果是false 系统会返回redigo: connection pool exhausted // 如果是true 会利用p 的ch 属性让线程等待 知道有连接释放出来 Wait bool // 连接最长生存时间 如果超过时间会被从链表中删除 MaxConnLifetime time.Duration // 判断ch 是否被初始化了 chInitialized uint32 // set to 1 when field ch is initialized // 锁 mu sync.Mutex // mu protects the following fields closed bool // set to true when the pool is closed. active int // the number of open connections in the pool ch chan struct{} // limits open connections when p.Wait is true // 存放闲置连接的链表 idle idleList // idle connections // 等待获取连接的数量 waitCount int64 // total number of connections waited for. waitDuration time.Duration // total time waited for new connections.}// 连接池中的具体连接对象type conn struct { // 锁 mu sync.Mutex pending int err error // http 包中的conn对象 conn net.Conn // 读入过期时间 readTimeout time.Duration // bufio reader对象 用于读取redis服务返回的结果 br *bufio.Reader // 写入过期时间 writeTimeout time.Duration // bufio writer对象 带buf 用于往服务端写命令 bw *bufio.Writer // Scratch space for formatting argument length. // '*' or '$', length, "\r\n" lenScratch [32]byte // Scratch space for formatting integers and floats. numScratch [40]byte}我们可以看到，其中有几个关键性的字段比如最大活动连接数、最大闲置连接数、闲置链接过期时间、连接生存时间等。
Pool 的Get Close方法
我们知道 连接池最重要的就是两个方法，一个是获取连接，一个是关闭连接。这个跟sync.Pool。我们来看一下代码：
GET：
func (p *Pool) get(ctx context.Context) (*poolConn, error) { // 处理是否需要等待 pool Wait如果是true 则等待连接释放 var waited time.Duration if p.Wait && p.MaxActive > 0 { // 重新初始化pool的ch channel p.lazyInit() // wait indicates if we believe it will block so its not 100% accurate // however for stats it should be good enough. wait := len(p.ch) == 0 var start time.Time if wait { start = time.Now() } // 获取pool 的ch通道，一旦有连接被close 则可以继续返回连接 if ctx == nil { <-p.ch } else { select { case <-p.ch: case <-ctx.Done(): return nil, ctx.Err() } } if wait { waited = time.Since(start) } } p.mu.Lock() // 等待数量加1 增加等待时间 if waited > 0 { p.waitCount++ p.waitDuration += waited } // Prune stale connections at the back of the idle list. // 删除链表尾部的陈旧连接，删除超时的连接 // 连接close之后，连接会回到pool的idle(闲置)链表中 if p.IdleTimeout > 0 { n := p.idle.count for i := 0; i < n && p.idle.back != nil && p.idle.back.t.Add(p.IdleTimeout).Before(nowFunc()); i++ { pc := p.idle.back p.idle.popBack() p.mu.Unlock() pc.c.Close() p.mu.Lock() p.active-- } } // Get idle connection from the front of idle list. // 获取链表空闲连接 拿链表第一个 for p.idle.front != nil { pc := p.idle.front p.idle.popFront() p.mu.Unlock() // 调用验证函数如果返回错误不为nil 关闭连接拿下一个 // 判断连接生存时间 大于生存时间则关闭拿下一个 if (p.TestOnBorrow == nil || p.TestOnBorrow(pc.c, pc.t) == nil) && (p.MaxConnLifetime == 0 || nowFunc().Sub(pc.created) < p.MaxConnLifetime) { return pc, nil } pc.c.Close() p.mu.Lock() p.active-- } // Check for pool closed before dialing a new connection. // 判断连接池是否被关闭 如果关闭则解锁报错 if p.closed { p.mu.Unlock() return nil, errors.New("redigo: get on closed pool") } // Handle limit for p.Wait == false. // 如果活动连接大于最大连接解锁 返回错误 if !p.Wait && p.MaxActive > 0 && p.active >= p.MaxActive { p.mu.Unlock() return nil, ErrPoolExhausted } // 如果在链表中没有获取到可用的连接 并添加active数量添加 p.active++ p.mu.Unlock() c, err := p.dial(ctx) // 如果调用失败 则减少active数量 if err != nil { c = nil p.mu.Lock() p.active-- if p.ch != nil && !p.closed { p.ch <- struct{}{} } p.mu.Unlock() } // 创建连接 设置创建时间 return &poolConn{c: c, created: nowFunc()}, err}Put：
// 关闭方法func (ac *activeConn) Close() error { pc := ac.pc if pc == nil { return nil } ac.pc = nil // 判断连接的状态 发送取消事务 取消watch if ac.state&connectionMultiState != 0 { pc.c.Send("DISCARD") ac.state &^= (connectionMultiState | connectionWatchState) } else if ac.state&connectionWatchState != 0 { pc.c.Send("UNWATCH") ac.state &^= connectionWatchState } if ac.state&connectionSubscribeState != 0 { pc.c.Send("UNSUBSCRIBE") pc.c.Send("PUNSUBSCRIBE") // To detect the end of the message stream, ask the server to echo // a sentinel value and read until we see that value. sentinelOnce.Do(initSentinel) pc.c.Send("ECHO", sentinel) pc.c.Flush() for { p, err := pc.c.Receive() if err != nil { break } if p, ok := p.([]byte); ok && bytes.Equal(p, sentinel) { ac.state &^= connectionSubscribeState break } } } pc.c.Do("") // 把连接放入链表 ac.p.put(pc, ac.state != 0 || pc.c.Err() != nil) return nil}// 将连接 重新放入限制链表func (p *Pool) put(pc *poolConn, forceClose bool) error { p.mu.Lock() if !p.closed && !forceClose { pc.t = nowFunc() p.idle.pushFront(pc) if p.idle.count > p.MaxIdle { pc = p.idle.back p.idle.popBack() } else { pc = nil } } if pc != nil { p.mu.Unlock() pc.c.Close() p.mu.Lock() p.active-- } // 如果连接的ch 不为空 并且连接池没有关闭 则给channel中输入一个struct{}{} // 如果在连接打到最大活动数量之后 再获取连接并且pool的Wait为ture 会阻塞线程等待返回连接 if p.ch != nil && !p.closed { p.ch <- struct{}{} } p.mu.Unlock() return nil}总结
整个Pool整体流程，我大概画了一个图。
从初始化 =》获取 -》创建连接 =》返回连接 =》关闭连接 =》
其中还有一条线是Pool.Wait = true 会一直阻塞 一直到有连接Close 释放活动连接数 线程被唤醒返回闲置的连接
其实大部分的连接池都是类似的流程，比如goroutine，redis。