深入解析 go-redis v9:Redis 官方 Go 客户端的 RESP 协议、连接池与 Hook 架构

import (
    "context"
    "fmt"
    "github.com/redis/go-redis/v9"
)

var ctx = context.Background()

func main() {
    rdb := redis.NewClient(&redis.Options{
        Addr:     "localhost:6379",
        Password: "",
        DB:       0,
    })
    defer rdb.Close()

    rdb.Set(ctx, "key", "value", 0)
    val, _ := rdb.Get(ctx, "key").Result()
    fmt.Println(val)
}

┌──────────────────────────────────────────────────────────────────────┐
│  应用代码                                                            │
│  rdb.Set(ctx, key, val, 0)  →  *redis.StatusCmd                     │
│  rdb.Get(ctx, key)          →  *redis.StringCmd                     │
└────────────────────┬─────────────────────────────────────────────────┘
                     │
┌────────────────────▼─────────────────────────────────────────────────┐
│  cmdable 父类(redis.go:353)                                       │
│  所有 Cmdable 接口的 default impl(Set/Get/SetNX/...)                 │
└────────────────────┬─────────────────────────────────────────────────┘
                     │
�────────────────────▼─────────────────────────────────────────────────┐
│  baseClient(redis.go:353) — 真正的"客户端"父类                     │
│  ── Hooks / Process / _process / Pool / Retries ──                 │
└────────────────────┬─────────────────────────────────────────────────┘
                     │
┌────────────────────▼─────────────────────────────────────────────────┐
│  internal/pool(10819 行)                                         │
│  ── Pool / Conn / wantConn / hooks / pubsub / pool_sticky ──     │
└────────────────────┬─────────────────────────────────────────────────┘
                     │
┌────────────────────▼─────────────────────────────────────────────────┐
│  internal/proto(4306 行)                                         │
│  ── Reader(1064) / Writer(242) / redis_errors(539) / scan(185) ── │
└────────────────────┬─────────────────────────────────────────────────�
                     │
┌────────────────────▼─────────────────────────────────────────────────┐
│  net.Conn(TCP / Unix)                                            │
└──────────────────────────────────────────────────────────────────────┘

              并列:Cluster / Sentinel / Ring / Tx / PubSub
┌──────────────────────────────────────────────────────────────────────┐
│  Client / *ClusterClient / *Sentinel / *Ring / *Tx / *PubSub      │
│  ── 都内嵌 baseClient,通过不同 Pool 实现 ──                        │
└──────────────────────────────────────────────────────────────────────┘

type baseClient struct {
    opt            *Options
    initOnce       sync.Once
    hooksMixin     hooksMixin
    pool           pool.Pooler
    dialTimeout    time.Duration
    readTimeout    time.Duration
    writeTimeout   time.Duration
    ctx            context.Context
    cancel         context.CancelFunc

    // Reusable command slice for pipelined writes, currently used by
    // ProcessPipeline and Pipelined.
    cmds           []Cmder
    cmdsBuf        []Cmder  // 复用 buffer

    onClose        *onCloseHooks

    // Client-side caching
    cscPoolHook    csc.PoolHook
    cscActive      bool
    himport        *himport.Registry

    // Push notification processor(RESP3 push messages)
    pushProcessor  push.Processor

    // Optimistic mutex for Options clone during init
    optMu          sync.Mutex
}

type Hook interface {
    DialHook(next DialHook) DialHook
    ProcessHook(next ProcessHook) ProcessHook
    ProcessPipelineHook(next ProcessPipelineHook) ProcessPipelineHook
}

type hooksMixin struct {
    // hooksMu serializes writers(AddHook);readers never take it.
    hooksMu *sync.Mutex
    // state holds the immutable hook snapshot.Readers Load it lock-free;
    // writers publish a replacement copy-on-write under hooksMu.
    state *atomic.Pointer[hooksState]
}

// hooksState is an immutable snapshot of the hook configuration.Once stored
// in hooksMixin.state it is never mutated;AddHook builds a fresh copy.
type hooksState struct {
    slice   []Hook
    initial hooks
    current hooks
}

// process(ctx, cmd) → retry wrapper,带 OTel + hook
func (c *baseClient) process(ctx context.Context, cmd Cmder) error {
    err := c.hooksMixin.processHook(ctx, cmd)
    cmd.SetErr(err)
    return err
}

// _process(ctx, cmd, attempt) → 真正的重试循环 + 连接获取
func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int, capture *cscFetchCapture) (bool, *pool.Conn, error) {
    // 1. 获取连接(pinned to this operation)
    cn, err := c.getConn(ctx)
    if err != nil { return false, nil, err }

    // 2. retry policy 检查
    retry, err := c.shouldRetry(ctx, cmd, attempt)
    if err != nil || !retry {
        // 3. write + read(主路径)
        if err := c.withWriter(ctx, cn, cmd); err != nil { ... }
        if err := cn.WithReader(ctx, c.cmdTimeout(cmd), cmd); err != nil { ... }
        return ...
    }

    // 4. 重试
    ...
}

const (
    RespStatus    = '+' // +<string>\r\n
    RespError     = '-' // -<string>\r\n
    RespString    = '$' // $<length>\r\n<bytes>\r\n
    RespInt       = ':' // :<number>\r\n
    RespNil       = '_' // _\r\n
    RespFloat     = ',' // ,<floating-point-number>\r\n
    RespBool      = '#' // true: #t\r\n false: #f\r\n
    RespBlobError = '!' // !<length>\r\n<bytes>\r\n
    RespVerbatim  = '=' // =<length>\r\nFORMAT:<bytes>\r\n
    RespBigInt    = '(' // (<big number>\r\n
    RespArray     = '*' // *<len>\r\n...
    RespMap       = '%' // %<len>\r\n(key)\r\n(value)\r\n...
    RespSet       = '~' // ~<len>\r\n...
    RespAttr      = '|' // |<len>\r\n(key)\r\n(value)\r\n... + command reply
    RespPush      = '>' // ><len>\r\n...
)

const DefaultBufferSize = 32 * 1024  // 32 KiB 默认 buffer

type Reader struct {
    rd *bufio.Reader
}

func NewReader(rd io.Reader) *Reader {
    return &Reader{
        rd: bufio.NewReaderSize(rd, DefaultBufferSize),
    }
}

func (r *Reader) PeekReplyType() (byte, error) {
    b, err := r.rd.Peek(1)  // ← 看一眼下一字节,不消费
    if err != nil { return 0, err }
    if b[0] == RespAttr {
        // ← attribute 先消费掉
        if err = r.DiscardNext(); err != nil { ... }
        b, err = r.rd.Peek(1)
    }
    return b[0], err
}

type Pool struct {
    opt *Options

    dialErrorsNum     uint32     // 累计 dial 失败数
    dialErrorsLock    sync.Mutex

    queue             chan struct{}    // 待 dial 信号
    connsMu           sync.Mutex
    idleConns         map[string]*Conn  // addr → idle conn
    staleConns        []*Conn
    badConns          []*Conn           // 检测到 bad 的 conn

    pubsub *pubsub.Pubsub              // pubsub 专用 pool

    // Hooks
    dialHook             func(ctx context.Context, network, addr string) (net.Conn, error)
    onGet                func(ctx context.Context, cn *Conn)
    onPut                func(ctx context.Context, cn *Conn)
    onClose              func(ctx context.Context, cn *Conn)
    onRemove             func(ctx context.Context, cn *Conn, err error)

    // Stat
    idleConnsLen   atomic.Int64
    staleConnsLen  atomic.Int64

    _closed atomic.Bool

    // metric callbacks(12 种)
    metricConnectionCreateTimeCallback ...
}

func (p *Pool) getConn(ctx context.Context) (*Conn, error) {
    // 1. 检查是否 closed
    if p._closed.Load() { return nil, ErrClosed }

    // 2. 检查 MaxActiveConns(限流)
    if p.opt.MaxActiveConns > 0 {
        // ...wait...
    }

    // 3. 先尝试从 idleConns pop
    cn, err := p.popIdleConn(ctx)
    if err != nil { return nil, err }
    if cn != nil {
        // 3a. 测试连接是否可用(避免 idle 太久的 dead conn)
        if err := p.checkBadConn(cn); err != nil {
            p.removeConn(cn, err)
            // 重试或失败
        }
        // 3b. 触发 onGet hook + metric
        p.onGet(ctx, cn)
        return cn, nil
    }

    // 4. dial 新连接(可能需要等 dial queue)
    return p.dialConn(ctx)
}

// pool_sticky.go
type stickyConnPool struct {
    pool       *Pool
    stickyLock sync.Mutex
    sticky     *Conn          // 持有 conn,不归还
    stickyUsed int
}

func (p *stickyConnPool) getConn(ctx context.Context) (*Conn, error) {
    p.stickyLock.Lock()
    cn := p.sticky
    p.stickyUsed++
    p.stickyLock.Unlock()
    if cn != nil { return cn, nil }
    // 从 parent pool 借一个
    return p.pool.getConn(ctx)
}

// want_conn.go
type wantConn struct {
    ctx    context.Context
    cancel context.CancelFunc
    cn     *Conn
    err    error
    ready  chan struct{}      // ← ready 时关闭
}

func (p *Pool) getConnAsync(ctx context.Context, addr string) (*wantConn, error) {
    w := &wantConn{
        ready: make(chan struct{}),
    }
    w.ctx, w.cancel = context.WithContext(ctx)
    // 入队
    p.queue <- struct{}{}
    go p.waitForConn(w)
    return w, nil
}

// otel.go
type otelHook struct{}

func (otelHook) DialHook(next DialHook) DialHook {
    return func(ctx context.Context, network, addr string) (net.Conn, error) {
        // 1. 启动 trace span
        tracer := otel.Tracer("redis-go")
        ctx, span := tracer.Start(ctx, "dial", trace.WithAttributes(...))
        defer span.End()

        // 2. 调下层
        cn, err := next(ctx, network, addr)
        if err != nil { span.RecordError(err); span.SetStatus(codes.Error, err.Error()) }
        return cn, err
    }
}

func (otelHook) ProcessHook(next ProcessHook) ProcessHook {
    return func(ctx context.Context, cmd Cmder) error {
        // 1. 启动 span
        ctx, span := tracer.Start(ctx, "redis-"+cmd.Name(), ...)

        // 2. 调下层
        err := next(ctx, cmd)

        // 3. 记录 metric
        span.SetAttributes(attribute.String("redis.command", cmd.Name()))
        if err != nil {
            span.RecordError(err)
            span.SetStatus(codes.Error, err.Error())
        }
        return err
    }
}

type pipelineQueue struct {
    mu    sync.Mutex
    cmds  []Cmder
    flush chan struct{}
}

func (p *pipelineQueue) start(ctx context.Context) {
    ticker := time.NewTicker(p.opts.AutoPipelineMaxDelay)
    defer ticker.Stop()
    for {
        select {
        case <-ctx.Done(): return
        case <-p.flush:
            p.flushNow()
        case <-ticker.C:
            p.flushNow()
        }
    }
}

$ find . -name "*.go" -not -path "./.git*" -not -name "*_test.go" | xargs wc -l | tail -1
  68,058 total

非测试代码 top 15:
command.go                       9,714 行
search_commands.go               4,038 行
osscluster.go                    3,072 行
redis.go                         2,659 行
autopipeline.go                  2,609 行
timeseries_commands.go           1,547 行
probabilistic.go                 1,499 行
sentinel.go                      1,338 行
ring.go                          1,028 行
osscluster_router.go             1,008 行
options.go                         999 行
string_commands.go                 927 行
csc_integration.go                 922 行
commands.go                        918 行
search_builders.go                 883 行

子模块规模:
internal/pool/                  10,819 行
internal/proto/                  4,306 行
internal/otel/                     900 行
internal/hscan/                    600 行
internal/hashtag/                  ~250 行
internal/routing/                  ~500 行
internal/util/                     ~300 行
internal/auth/streaming/           ~400 行
internal/log.go                    ~200 行
internal/semaphore.go              ~50 行

csc_*                            ~14,000 行(Client-Side Caching 整套)
maintnotifications/              ~2,000 行