Redigo Pool 最重要的结构
type Pool struct {
// 真正获取跟redis-server连接的函数, 必填参数
Dial func() (Conn, error)
// 这是个可选参数, 用于在从 pool 获取连接时, 检查这个连接是否正常使用. 所以这个参数一般是必填的
TestOnBorrow func(c Conn, t time.Time) error
// 最多有多少个空闲连接保留, 一般必填
MaxIdle int
// 最多有多少活跃的连接数, 一般必填
MaxActive int
// 空闲连接最长空闲时间, 一般必填
IdleTimeout time.Duration
// Pool 的活跃的连接数达到 MaxActive, 如果 Wait 为 true,
// 那么 Get() 将要等待一个连接放到 Pool中, 才会返回一个连接给使用方
Wait bool
// 设置连接最大存活时间
MaxConnLifetime time.Duration
chInitialized uint32 // set to 1 when field ch is initialized
mu sync.Mutex // mu protects the following fields
closed bool // 设置 Pool 是否关闭
active int // 当前 Pool 的活跃连接数
ch chan struct{} // 配合 Wait 为 true 使用
idle idleList // 空闲队列
}
Redigo 第二重要的结构: idleList
idleList 是个双向链表. 实现很简单. 只有三个方法: pushFront, popFront, popBack
初始化 idlelist
type idleList struct {
count int
front, back *poolConn
}
使用 Pool 不会明确的初始化 idle, 故当初始化&Pool{....}后, idle 就是默认值. 即: count = 0, front = nil, back = nil. 如下图:
头部插入
idleList 只提供 pushFront 方法. 将 idleList 的 front 指针, 指向新的 Conn. 然后将新 Conn 与 之前节点连接即可, 如下图
从头部删除
从尾部删除
Conn
整个 Pool 有两个 Conn: activeConn, errorConn, 他们都实现了 redis.Conn 接口.
type Conn interface {
// Close closes the connection.
Close() error
// Err returns a non-nil value when the connection is not usable.
Err() error
// Do sends a command to the server and returns the received reply.
Do(commandName string, args ...interface{}) (reply interface{}, err error)
// Send writes the command to the client's output buffer.
Send(commandName string, args ...interface{}) error
// Flush flushes the output buffer to the Redis server.
Flush() error
// Receive receives a single reply from the Redis server
Receive() (reply interface{}, err error)
}
不过 errorConn 实现的所有的函数都会返回 err. 所以我们在使用 Get() 获取链接时, 尽量要去判断拿到的连接是否可用(不过这也不是绝对, 我们也可以在使用 Conn 函数的时候去判断).
activeConn 一般来说是可用的连接, 我们也可以通过 activeConn.Err() 来判断获取的连接是否可用.
activeConn 的函数都是调用 redis.Conn 的函数实现的. 唯一不同的地方在于, 所有的命令都会执行LookupCommandInfo这个函数.
func LookupCommandInfo(commandName string) CommandInfo {
if ci, ok := commandInfos[commandName]; ok {
return ci
}
return commandInfos[strings.ToUpper(commandName)]
}
func (ac *activeConn) Do(commandName string, args ...interface{}) (reply interface{}, err error) {
pc := ac.pc
if pc == nil {
return nil, errConnClosed
}
ci := internal.LookupCommandInfo(commandName)
ac.state = (ac.state | ci.Set) &^ ci.Clear
return pc.c.Do(commandName, args...)
}
以Multi, EXEC举例:
使用 Pool 获取到的连接, 发送 Multi完, 由于某种原因导致程序执行了 defer activeConn.Close(), 由于 redis-server 没有得到Exec, 那么接下来的命令 redis-server 都会把其当成事务的一部分. 所以通过 LookupCommandInfo 函数能够计算当前某条命令执行后 activeConn 的当前状态, 当执行到 activeConn.Close() 时发现还没有发送 EXEC, 那么就会发送 DISCARD 命令来将事务取消.
LookupCommandInfo 支持的命令有: WATCH, UNWATCH, MULTI, EXEC, DISCARD, PSUBSCRIBE, SUBSCRIBE, MONITOR
Redigo 灵魂函数 -- get()
get() 主要提供给 Get() 函数调用, 是从 Pool 中获取连接, Get()是我们使用者调用的函数. 可以看到 Get() 由 errorConn, activeConn 组成.
func (p *Pool) Get() Conn {
pc, err := p.get(nil)
if err != nil {
return errorConn{err}
}
return &activeConn{p: p, pc: pc}
}
Wait 配合 MaxActive 使用, 来保证 Get() 将要等待一个连接放到 Pool中, 才会返回一个连接给使用方
if p.Wait && p.MaxActive > 0 {
p.lazyInit()
if ctx == nil {
<-p.ch // <--- 这里会阻塞
} else {
// 下面的代码不需要关于, Get() 传递下来的参数是 gnil
select {
case <-p.ch:
case <-ctx.Done():
return nil, ctx.Err()
}
}
}
设置 Wait 为 true 并且 MaxActive 设置有最大数时, 如果时第一次获取 Get(), 那么会调用lazyInit()进行初始化.
初始化一个 MaxActive 的 channel. 利用 channel 来保证当 Get() 获取到的连接数大于 MaxActive时, 阻塞 Get() 函数, 直到有连接使用完毕放入到 Pool 中.
循环idleList, 关闭空闲队列中连接时长大于 IdleTimeout 的连接
// Prune stale connections at the back of the idle list.
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--
}
}
这段代码会遍历整个 idleList, 从尾部拿出 activeConn, activeConn.t 加上 IdleTimeout 时间, 跟当前时间比较. 如果比 time.Now() 小, 则从 idleList 的尾部 pop 这个conn. 同时关闭这个 activeConn, 让 Pool.active--
不过这段代码看上去比较奇怪, 需要耐心去看. 可以做个变形
for i := 0; i < n ; i++ {
if p.idle.back != nil && p.idle.back.t.Add(p.IdleTimeout).Before(nowFunc()) {
pc := p.idle.back
p.idle.popBack()
p.mu.Unlock()
pc.c.Close()
p.mu.Lock()
p.active--
}
}
当 idleList 不为空时, 从头部获取 activeConn
for p.idle.front != nil {
pc := p.idle.front
p.idle.popFront()
p.mu.Unlock()
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--
}
这段代码比较简单.当 idleList 不为空时, 从头部 pop 出一个 activeConn, 还有另外两个功能:
- 通过
TestOnBorrow()函数判断当前连接是否能正常使用 - 通过
MaxConnLifetime参数判断这个连接是否在 MaxConnLifetime 内
判断 Pool 是否关闭
if p.closed {
p.mu.Unlock()
return nil, errors.New("redigo: get on closed pool")
}
当发现 Pool 被调用 Pool.Close() 关闭了, 那么这里就会返回 errors.New("redigo: get on closed pool")错误
顺带说一句在 `pool.go` 里面总共有两个 Close() 函数:
1. func (p *Pool) Close() error {...}
这个函数是关闭 redigo 连接池的. 理论上可以不调用. 如果确实不放心, 需要在 main.go 里面 `defer pool.Close()`
来调用
2. func (ac *activeConn) Close() error { ...}
这个函数是用来将从 Pool 中获取到的 activeConn 放回到 Pool 里面.
这个函数是我们需要频繁调用的函数. 如果程序里Get() 之后没有 Close(), 那么就会造成 redis 连接泄漏.
更严重的情况, 如果Wait, MaxActive 都没有设置, 那么你的程序就会将 redis 搞瘫痪, 这是很危险的
真正从 redis-server 获取连接
// Handle limit for p.Wait == false.
if !p.Wait && p.MaxActive > 0 && p.active >= p.MaxActive {
p.mu.Unlock()
return nil, ErrPoolExhausted
}
p.active++
p.mu.Unlock()
c, err := p.Dial()
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
- 判断 pool 的 active 是否达到 MaxActive
- 通过参数 p.Dial() 去 redis-server 获取连接
Redigo 灵魂函数 -- put()
put 函数主要提供给 activeConn.Close() 调用
Close() 函数就不在详细说明, 主要根据 activeConn 的 stat, 判断在关闭连接之前是否发送过WATCH, MULTI, PSUBSCRIBE, SUBSCRIBE, MONITOR 这些命令. 如果发送过就会把这些命令结束(具体原因上面已经说过)
func (p *Pool) put(pc *poolConn, forceClose bool) error {
p.mu.Lock()
// 判断 pool 是否关闭, 并且该命令是否需要强制关闭
if !p.closed && !forceClose {
pc.t = nowFunc()
// 将该 activeConn 压入 idleList 中
p.idle.pushFront(pc)
// 如果 idleList 的 count 已经大于 MaxIdle, 那么会将 idleList 的尾部的 activeConn pop 掉
if p.idle.count > p.MaxIdle {
pc = p.idle.back
p.idle.popBack()
} else {
pc = nil
}
}
// 如果是需要强制关闭或者是从尾部 pop 掉的 conn, 那么就会真正的关闭这个连接
if pc != nil {
p.mu.Unlock()
pc.c.Close()
p.mu.Lock()
p.active--
}
// 如果开启了 Wait = true, 那么往 channel 里面发送一个struct{}{}, 代表等待的客户端可以获取连接了
if p.ch != nil && !p.closed {
p.ch <- struct{}{}
}
p.mu.Unlock()
return nil
}
结尾
至此, redigo Pool 的源码基本都过了一遍. 为什么我会心血来潮把其源码读一遍呢?
思考下面几个问题:
- redigo 是否能够用于 codis?
- codis 的 golang 客户端如何实现 ?
- 如果不经过任何加工, 直接用 redigo 去访问 codis, 会出现什么样的问题?
这些问题就是促使我读redigo源码的原因, 后续文章我会一一解答这些问题
