Go Redigo 源码分析(二) 链接池

Redigo 链接池的使用

你们都知道go语言中的goroutine虽然消耗资源很小，而且是一个用户线程。可是goroutine也不是无限开的，因此咱们会有不少关于协程池的库，固然啊咱们本身也能够完成一些简单的携程池。redis也是相同的，redis的连接也是不推荐无限制的打开，不然会形成redis负荷加剧。
先看一下Redigo 中的链接池的使用

package main

import (
    "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使用链接池仍是很简单的步骤：

1. 建立链接池
2. 简单设置链接池的最大连接数等参数
3. 注入拨号函数（设置redis地址 端口号等）
4. 调用pool.Get() 获取链接
5. 使用链接Do函数请求redis
6. 关闭链接

源码

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。