redis

golang

发布日期: 2022-04-12

更新日期: 2025-12-05

文章字数: 4.9k

阅读时长: 25 分

go-redis

golang实现的redis客户端

call chain

调用分层

获取客户端
函数生成命令对象
命令对象转换为RESP协议
网络层发送协议
网络层接收协议
转换为命令响应
返回给调用者

pipeline调用内部逻辑

client.Pipelined() {
	client.processPipeline() {
		hooks.processPipeline(
			baseClient._generalProcessPipeline() {
				baseClient.withConn(
					baseClient.pipelineProcessCmds() {
						Conn.WithWriter(
							Writer.WriteArgs()
						)
						Conn.WithReader(
							Cmder.readReply()
						)
					}
				)
			}
		）
	}
}

---
title: go-redis 发起请求流程图
---
sequenceDiagram
    participant User
    participant Client
    participant Hooks
    participant BaseClient
    participant Conn
    participant Net
    participant Cmder

    User->>Client: Pipelined()
    activate Client
    Client->>Client: processPipeline()
    Client->>Hooks: hooks.processPipeline()
    activate Hooks
    Hooks->>BaseClient: _generalProcessPipeline()
    activate BaseClient
    BaseClient->>BaseClient: withConn()
    BaseClient->>BaseClient: pipelineProcessCmds()
    BaseClient->>Conn: WithWriter()
    activate Conn
    Conn->>Net: WriteArgs()
    Net->>Conn: WithReader()
    Conn->>Cmder: readReply()
	Cmder->>Conn: return
    deactivate Conn
	Conn->>BaseClient: return
    deactivate BaseClient
	BaseClient->>Hooks: return
    deactivate Hooks
	Hooks->>Client: return
    deactivate Client
	Client->>User: return

core object

执行单元：client，baseClient，pipeline，txPipeline
协议：Reader，Writer
连接池：connPool，singlePool，stickyPool

client

向用户提供对redis的操作

option

配置redis地址，超时，连接池等信息

// Options keeps the settings to setup redis connection.
type Options struct {
	// 网络模式
	Network string
	// 地址
	Addr string

	// 拨号器
	Dialer func(ctx context.Context, network, addr string) (net.Conn, error)

	// 连接hook
	OnConnect func(ctx context.Context, cn *Conn) error

	// 用户名 用于 redis ACL
	Username string
	// 密码
	Password string

	// 数据库
	DB int

	// 最大重试次数
	MaxRetries int
	// 重试最小间隔
	MinRetryBackoff time.Duration
	// 最大重试间隔
	MaxRetryBackoff time.Duration

	// 拨号超时
	DialTimeout time.Duration
	// 读超时
	ReadTimeout time.Duration
	// 写超时
	WriteTimeout time.Duration

	// 连接池类型
	// true for FIFO pool, false for LIFO pool.
	PoolFIFO bool
	// 最大连接数 默认 十倍runtime.GOMAXPROCS.
	PoolSize int
	// 最小空闲连接数
	MinIdleConns int
	// 连接最大存活时间
	MaxConnAge time.Duration
	// 获取连接超时时间
	PoolTimeout time.Duration
	// 空闲连接超时时间
	IdleTimeout time.Duration
	// 空闲连接检测频次
	IdleCheckFrequency time.Duration

	// 只读节点
	readOnly bool

	// TLS 配置
	TLSConfig *tls.Config

	// 限流器
	Limiter Limiter
}

// 默认值
func (opt *Options) init() {
	if opt.Addr == "" {
		opt.Addr = "localhost:6379"
	}
	if opt.Network == "" {
		if strings.HasPrefix(opt.Addr, "/") {
			opt.Network = "unix"
		} else {
			opt.Network = "tcp"
		}
	}
	if opt.DialTimeout == 0 {
		opt.DialTimeout = 5 * time.Second
	}
	if opt.Dialer == nil {
		opt.Dialer = func(ctx context.Context, network, addr string) (net.Conn, error) {
			netDialer := &net.Dialer{
				Timeout:   opt.DialTimeout,
				KeepAlive: 5 * time.Minute,
			}
			if opt.TLSConfig == nil {
				return netDialer.DialContext(ctx, network, addr)
			}
			return tls.DialWithDialer(netDialer, network, addr, opt.TLSConfig)
		}
	}
	if opt.PoolSize == 0 {
		opt.PoolSize = 10 * runtime.GOMAXPROCS(0)
	}
	switch opt.ReadTimeout {
	case -1:
		opt.ReadTimeout = 0
	case 0:
		opt.ReadTimeout = 3 * time.Second
	}
	switch opt.WriteTimeout {
	case -1:
		opt.WriteTimeout = 0
	case 0:
		opt.WriteTimeout = opt.ReadTimeout
	}
	if opt.PoolTimeout == 0 {
		opt.PoolTimeout = opt.ReadTimeout + time.Second
	}
	if opt.IdleTimeout == 0 {
		opt.IdleTimeout = 5 * time.Minute
	}
	if opt.IdleCheckFrequency == 0 {
		opt.IdleCheckFrequency = time.Minute
	}

	if opt.MaxRetries == -1 {
		opt.MaxRetries = 0
	} else if opt.MaxRetries == 0 {
		opt.MaxRetries = 3
	}
	switch opt.MinRetryBackoff {
	case -1:
		opt.MinRetryBackoff = 0
	case 0:
		opt.MinRetryBackoff = 8 * time.Millisecond
	}
	switch opt.MaxRetryBackoff {
	case -1:
		opt.MaxRetryBackoff = 0
	case 0:
		opt.MaxRetryBackoff = 512 * time.Millisecond
	}
}

hook

在执行时的钩子，通过钩子可以在执行redis操作时，额外执行用户自定义的操作，如链路追踪，消息打点

type Hook interface {
	BeforeProcess(ctx context.Context, cmd Cmder) (context.Context, error)
	AfterProcess(ctx context.Context, cmd Cmder) error

	BeforeProcessPipeline(ctx context.Context, cmds []Cmder) (context.Context, error)
	AfterProcessPipeline(ctx context.Context, cmds []Cmder) error
}

执行命令，单命令，批量命令，事务命令

func (hs hooks) process(ctx context.Context, cmd Cmder, fn func(context.Context, Cmder) error) error，带hook的处理单命令
func (hs hooks) processPipeline(ctx context.Context, cmds []Cmder, fn func(context.Context, []Cmder) error) error，带hook的处理多命令
func (hs hooks) processTxPipeline(ctx context.Context, cmds []Cmder, fn func(context.Context, []Cmder) error) error，带hook的事务处理多命令

baseClient

主要负责连接池管理和redis命令处理

type baseClient struct {
	opt      *Options
	connPool pool.Pooler

	onClose func() error // hook called when client is closed
}

管理连接

func (c *baseClient) newConn(ctx context.Context) (*pool.Conn, error)，新建连接
func (c *baseClient) getConn(ctx context.Context) (*pool.Conn, error)，从池中获取连接
func (c *baseClient) initConn(ctx context.Context, cn *pool.Conn) error，初始化连接
func (c *baseClient) releaseConn(ctx context.Context, cn *pool.Conn, err error)，释放连接
func (c *baseClient) withConn(ctx context.Context, fn func(context.Context, *pool.Conn) error) error，绑定连接执行fn

执行命令

func (c *baseClient) process(ctx context.Context, cmd Cmder) error，处理单命令
func (c *baseClient) processPipeline(ctx context.Context, cmds []Cmder) error，处理多个命令
func (c *baseClient) processTxPipeline(ctx context.Context, cmds []Cmder) error，事务处理多个命令
func (c *baseClient) pipelineProcessCmds(ctx context.Context, cn *pool.Conn, cmds []Cmder) (bool, error)，执行多个命令
func (c *baseClient) txPipelineProcessCmds(ctx context.Context, cn *pool.Conn, cmds []Cmder) (bool, error)，事务执行多个命令

Client

封装与用户的交互

type Client struct {
	*baseClient
	cmdable
	hooks
	ctx context.Context
}

执行命令

func (c *Client) Do(ctx context.Context, args ...interface{}) *Cmd，执行命令，手动拼接命令
func (c *Client) Process(ctx context.Context, cmd Cmder) error，执行命令
func (c *Client) processPipeline(ctx context.Context, cmds []Cmder) error，执行多个命令
func (c *Client) processTxPipeline(ctx context.Context, cmds []Cmder) error，事务执行多个命令
func (c *Client) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error)，管道执行fn
func (c *Client) Pipeline() Pipeliner，返回管道
func (c *Client) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error)，事务管道执行fn
func (c *Client) TxPipeline() Pipeliner，返回事务管道

Conn

单连接客户端，比Client多了些有状态的命令，用的比较少

type conn struct {
	baseClient
	cmdable
	statefulCmdable
	hooks // TODO: inherit hooks
}

type Conn struct {
	*conn
	ctx context.Context
}

执行命令

func (c *Conn) Process(ctx context.Context, cmd Cmder) error，执行命令
func (c *Conn) processPipeline(ctx context.Context, cmds []Cmder) error，执行多个命令
func (c *Conn) processTxPipeline(ctx context.Context, cmds []Cmder) error，事务执行命令
func (c *Conn) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error)，利用管道执行fn
func (c *Conn) Pipeline() Pipeliner，返回管道
func (c *Conn) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error)，利用事务管道执行fn
func (c *Conn) TxPipeline() Pipeliner，返回事务管道

Tx

事务执行命令，会监视所有操作的key

type Tx struct {
	baseClient
	cmdable
	statefulCmdable
	hooks
	ctx context.Context
}

执行命令

func (c *Tx) Process(ctx context.Context, cmd Cmder) error，执行命令
func (c *Tx) Close(ctx context.Context) error，关闭连接
func (c *Tx) Watch(ctx context.Context, keys ...string) *StatusCmd，监视keys
func (c *Tx) Unwatch(ctx context.Context, keys ...string) *StatusCmd，取消监视keys
func (c *Tx) Pipeline() Pipeliner，返回管道
func (c *Tx) Pipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error)，利用管道执行fn
func (c *Tx) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([]Cmder, error)，利用事务管道执行fn
func (c *Tx) TxPipeline() Pipeliner，返回事务管道

command

处理redis命令的文件

commands.go，用于描述redis命令，处理请求
command.go，用于描述代码层级的结构，处理响应

commands

commands.go为每个redis命令实现了一个或多个操作函数，针对每个命令返回值的不同，返回指定的响应结构体（即command.go中定义的结构），有些指令还会预处理参数方便用户使用。
时间格式化：

func formatMs(ctx context.Context, dur time.Duration) int64 {
	if dur > 0 && dur < time.Millisecond {
		internal.Logger.Printf(
			ctx,
			"specified duration is %s, but minimal supported value is %s - truncating to 1ms",
			dur, time.Millisecond,
		)
		return 1
	}
	return int64(dur / time.Millisecond)
}

func formatSec(ctx context.Context, dur time.Duration) int64 {
	if dur > 0 && dur < time.Second {
		internal.Logger.Printf(
			ctx,
			"specified duration is %s, but minimal supported value is %s - truncating to 1s",
			dur, time.Second,
		)
		return 1
	}
	return int64(dur / time.Second)
}

参数格式化：

func appendArgs(dst, src []interface{}) []interface{} {
	if len(src) == 1 {
		return appendArg(dst, src[0])
	}

	dst = append(dst, src...)
	return dst
}

func appendArg(dst []interface{}, arg interface{}) []interface{} {
	switch arg := arg.(type) {
	case []string:
		for _, s := range arg {
			dst = append(dst, s)
		}
		return dst
	case []interface{}:
		dst = append(dst, arg...)
		return dst
	case map[string]interface{}:
		for k, v := range arg {
			dst = append(dst, k, v)
		}
		return dst
	case map[string]string:
		for k, v := range arg {
			dst = append(dst, k, v)
		}
		return dst
	default:
		return append(dst, arg)
	}
}

command

command.go定义了Cmder接口用于golang和redis之间的交互，主要作用是将golang对象转为RESP协议发送给redis。

用户读取信息的操作接口

type Cmder interface {
	Name() string         // redis 命令
	FullName() string     // 多个单词的redis命令
	Args() []interface{}  // 命令参数列表
	String() string       // 格式化
	stringArg(int) string // 格式化参数
	firstKeyPos() int8
	setFirstKeyPos(int8)

	readTimeout() *time.Duration      // 读超时
	readReply(rd *proto.Reader) error // 读取值

	SetErr(error)
	Err() error
}

最基本操作单元

type baseCmd struct {
	ctx    context.Context
	args   []interface{} // 命令参数 cmd args...
	err    error
	keyPos int8

	_readTimeout *time.Duration
}

参数

func (cmd *baseCmd) Name() string，返回第一个参数，一般第一个参数都是操作命令
func (cmd *baseCmd) FullName() string，返回操作命令，redis有多个单词组成的命令
func (cmd *baseCmd) Args() []interface{}，返回参数列表
func (cmd *baseCmd) stringArg(pos int) string，以字符串形式返回指定位置的参数，利用断言机制
错误
func (cmd *baseCmd) SetErr(e error)，设置错误
func (cmd *baseCmd) Err() error，返回错误
读超时
func (cmd *baseCmd) readTimeout() *time.Duration，返回读超时
func (cmd *baseCmd) setReadTimeout(d time.Duration)，设置读超时

type Cmd struct {
	baseCmd

	val interface{}
}

序列化

func (cmd *Cmd) String() string，格式化Cmd，即格式化参数和返回值

返回值

func (cmd *Cmd) SetVal(val interface{})，设置返回值
func (cmd *Cmd) Val() interface{}，获取返回值
func (cmd *Cmd) Result() (interface{}, error)，获取返回值和错误

还实现了将返回值转换为指定类型的函数，包括：Text，Int，Int64，Uint64，Float32，Float64，Bool，Slice，StringSlice，Int64Slice，Uint64Slice，Float32Slice，Float64Slice，BoolSlice

读取返回值

func (cmd *Cmd) readReply(rd *proto.Reader) (err error) {
	cmd.val, err = rd.ReadReply(sliceParser)
	return err
}

// sliceParser implements proto.MultiBulkParse.
func sliceParser(rd *proto.Reader, n int64) (interface{}, error) {
	vals := make([]interface{}, n)
	for i := 0; i < len(vals); i++ {
		v, err := rd.ReadReply(sliceParser)
		if err != nil {
			if err == Nil {
				vals[i] = nil
				continue
			}
			if err, ok := err.(proto.RedisError); ok {
				vals[i] = err
				continue
			}
			return nil, err
		}
		vals[i] = v
	}
	return vals, nil
}

SliceCmd

type SliceCmd struct {
	baseCmd

	val []interface{}
}

// Scan scans the results from the map into a destination struct. The map keys
// are matched in the Redis struct fields by the `redis:"field"` tag.
func (cmd *SliceCmd) Scan(dst interface{}) error {
	if cmd.err != nil {
		return cmd.err
	}

	// Pass the list of keys and values.
	// Skip the first two args for: HMGET key
	var args []interface{}
	if cmd.args[0] == "hmget" {
		args = cmd.args[2:]
	} else {
		// Otherwise, it's: MGET field field ...
		args = cmd.args[1:]
	}

	return hscan.Scan(dst, args, cmd.val)
}

func (cmd *SliceCmd) readReply(rd *proto.Reader) error {
	v, err := rd.ReadArrayReply(sliceParser)
	if err != nil {
		return err
	}
	cmd.val = v.([]interface{})
	return nil
}

StatusCmd

type StatusCmd struct {
	baseCmd

	val string
}

func (cmd *StatusCmd) readReply(rd *proto.Reader) (err error) {
	cmd.val, err = rd.ReadString()
	return err
}

IntCmd

type IntCmd struct {
	baseCmd

	val int64
}

func (cmd *IntCmd) readReply(rd *proto.Reader) (err error) {
	cmd.val, err = rd.ReadIntReply()
	return err
}

IntSliceCmd

type IntSliceCmd struct {
	baseCmd

	val []int64
}

func (cmd *IntSliceCmd) readReply(rd *proto.Reader) error {
	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
		cmd.val = make([]int64, n)
		for i := 0; i < len(cmd.val); i++ {
			num, err := rd.ReadIntReply()
			if err != nil {
				return nil, err
			}
			cmd.val[i] = num
		}
		return nil, nil
	})
	return err
}

DurationCmd

type DurationCmd struct {
	baseCmd

	val       time.Duration
	precision time.Duration
}

func (cmd *DurationCmd) readReply(rd *proto.Reader) error {
	n, err := rd.ReadIntReply()
	if err != nil {
		return err
	}
	switch n {
	// -2 if the key does not exist
	// -1 if the key exists but has no associated expire
	case -2, -1:
		cmd.val = time.Duration(n)
	default:
		cmd.val = time.Duration(n) * cmd.precision
	}
	return nil
}

TimeCmd

type TimeCmd struct {
	baseCmd

	val time.Time
}

func (cmd *TimeCmd) readReply(rd *proto.Reader) error {
	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
		if n != 2 {
			return nil, fmt.Errorf("got %d elements, expected 2", n)
		}

		sec, err := rd.ReadInt()
		if err != nil {
			return nil, err
		}

		microsec, err := rd.ReadInt()
		if err != nil {
			return nil, err
		}

		cmd.val = time.Unix(sec, microsec*1000)
		return nil, nil
	})
	return err
}

BoolCmd

type BoolCmd struct {
	baseCmd

	val bool
}

func (cmd *BoolCmd) readReply(rd *proto.Reader) error {
	v, err := rd.ReadReply(nil)
	// `SET key value NX` returns nil when key already exists. But
	// `SETNX key value` returns bool (0/1). So convert nil to bool.
	if err == Nil {
		cmd.val = false
		return nil
	}
	if err != nil {
		return err
	}
	switch v := v.(type) {
	case int64:
		cmd.val = v == 1
		return nil
	case string:
		cmd.val = v == "OK"
		return nil
	default:
		return fmt.Errorf("got %T, wanted int64 or string", v)
	}
}

StringCmd

type StringCmd struct {
	baseCmd

	val string
}

func (cmd *StringCmd) readReply(rd *proto.Reader) (err error) {
	cmd.val, err = rd.ReadString()
	return err
}

还封装了Bytes，Bool，Int，Int64，Uint64，Float32，Float64，Time，Scan

FloatCmd

type FloatCmd struct {
	baseCmd

	val float64
}

func (cmd *FloatCmd) readReply(rd *proto.Reader) (err error) {

	cmd.val, err = rd.ReadFloatReply()
	return err
}

FloatSliceCmd

type FloatSliceCmd struct {
	baseCmd

	val []float64
}

func (cmd *FloatSliceCmd) readReply(rd *proto.Reader) error {
	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
		cmd.val = make([]float64, n)
		for i := 0; i < len(cmd.val); i++ {
			switch num, err := rd.ReadFloatReply(); {
			case err == Nil:
				cmd.val[i] = 0
			case err != nil:
				return nil, err
			default:
				cmd.val[i] = num
			}
		}
		return nil, nil
	})
	return err
}

StringSliceCmd

type StringSliceCmd struct {
	baseCmd

	val []string
}

func (cmd *StringSliceCmd) readReply(rd *proto.Reader) error {
	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
		cmd.val = make([]string, n)
		for i := 0; i < len(cmd.val); i++ {
			switch s, err := rd.ReadString(); {
			case err == Nil:
				cmd.val[i] = ""
			case err != nil:
				return nil, err
			default:
				cmd.val[i] = s
			}
		}
		return nil, nil
	})
	return err
}

BoolSliceCmd

type BoolSliceCmd struct {
	baseCmd

	val []bool
}

func (cmd *BoolSliceCmd) readReply(rd *proto.Reader) error {
	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
		cmd.val = make([]bool, n)
		for i := 0; i < len(cmd.val); i++ {
			n, err := rd.ReadIntReply()
			if err != nil {
				return nil, err
			}
			cmd.val[i] = n == 1
		}
		return nil, nil
	})
	return err
}

StringStringMapCmd

type StringStringMapCmd struct {
	baseCmd

	val map[string]string
}

func (cmd *StringStringMapCmd) readReply(rd *proto.Reader) error {
	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
		cmd.val = make(map[string]string, n/2)
		for i := int64(0); i < n; i += 2 {
			key, err := rd.ReadString()
			if err != nil {
				return nil, err
			}

			value, err := rd.ReadString()
			if err != nil {
				return nil, err
			}

			cmd.val[key] = value
		}
		return nil, nil
	})
	return err
}

func (cmd *StringStringMapCmd) Scan(dest interface{}) error {
	if cmd.err != nil {
		return cmd.err
	}

	strct, err := hscan.Struct(dest)
	if err != nil {
		return err
	}

	for k, v := range cmd.val {
		if err := strct.Scan(k, v); err != nil {
			return err
		}
	}

	return nil
}

StringIntMapCmd

type StringIntMapCmd struct {
	baseCmd

	val map[string]int64
}

func (cmd *StringIntMapCmd) readReply(rd *proto.Reader) error {
	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
		cmd.val = make(map[string]int64, n/2)
		for i := int64(0); i < n; i += 2 {
			key, err := rd.ReadString()
			if err != nil {
				return nil, err
			}

			n, err := rd.ReadIntReply()
			if err != nil {
				return nil, err
			}

			cmd.val[key] = n
		}
		return nil, nil
	})
	return err
}

StringStructMapCmd

type StringStructMapCmd struct {
	baseCmd

	val map[string]struct{}
}

func (cmd *StringStructMapCmd) readReply(rd *proto.Reader) error {
	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
		cmd.val = make(map[string]struct{}, n)
		for i := int64(0); i < n; i++ {
			key, err := rd.ReadString()
			if err != nil {
				return nil, err
			}
			cmd.val[key] = struct{}{}
		}
		return nil, nil
	})
	return err
}

ZSliceCmd

type ZSliceCmd struct {
	baseCmd

	val []Z
}

func (cmd *ZSliceCmd) readReply(rd *proto.Reader) error {
	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
		cmd.val = make([]Z, n/2)
		for i := 0; i < len(cmd.val); i++ {
			member, err := rd.ReadString()
			if err != nil {
				return nil, err
			}

			score, err := rd.ReadFloatReply()
			if err != nil {
				return nil, err
			}

			cmd.val[i] = Z{
				Member: member,
				Score:  score,
			}
		}
		return nil, nil
	})
	return err
}

ZWithKeyCmd

type ZWithKeyCmd struct {
	baseCmd

	val *ZWithKey
}

func (cmd *ZWithKeyCmd) readReply(rd *proto.Reader) error {
	_, err := rd.ReadArrayReply(func(rd *proto.Reader, n int64) (interface{}, error) {
		if n != 3 {
			return nil, fmt.Errorf("got %d elements, expected 3", n)
		}

		cmd.val = &ZWithKey{}
		var err error

		cmd.val.Key, err = rd.ReadString()
		if err != nil {
			return nil, err
		}

		cmd.val.Member, err = rd.ReadString()
		if err != nil {
			return nil, err
		}

		cmd.val.Score, err = rd.ReadFloatReply()
		if err != nil {
			return nil, err
		}

		return nil, nil
	})
	return err
}

ScanCmd

type ScanCmd struct {
	baseCmd

	page   []string
	cursor uint64

	process cmdable
}

func (cmd *ScanCmd) readReply(rd *proto.Reader) (err error) {
	cmd.page, cmd.cursor, err = rd.ReadScanReply()
	return err
}

func (cmd *ScanCmd) Iterator() *ScanIterator {
	return &ScanIterator{
		cmd: cmd,
	}
}

proto

详细文档：见RESP协议文档

const (
	ErrorReply  = '-' // 简单错误   -<string>\r\n
	StatusReply = '+' // 简单字符串 +<string>\r\n
	IntReply    = ':' // 数字      :<number>\r\n
	StringReply = '$' // 二进制数据 $<length>\r\n<bytes>\r\n
	ArrayReply  = '*' // 数组      *<elements number>\r\n... numelements other types ...
)

request

用于发送RESP字节流协议

type writer interface {
	io.Writer
	io.ByteWriter
	// io.StringWriter
	WriteString(s string) (n int, err error)
}

type Writer struct {
	writer

	lenBuf []byte
	numBuf []byte
}

golang数据转为RESP协议

func (w *Writer) WriteArgs(args []interface{}) error，写入多个基本类型
func (w *Writer) writeLen(n int) error，写入长度
func (w *Writer) WriteArg(v interface{}) error，写入基本类型
func (w *Writer) bytes(b []byte) error，写入字节流
func (w *Writer) string(s string) error，写入字符串
func (w *Writer) uint(n uint64) error，写入uint
func (w *Writer) int(n int64) error，写入int
func (w *Writer) float(f float64) error，写入浮点
func (w *Writer) crlf() error，写入回车换行

reply

用于接收redis返回的数据

type MultiBulkParse func(*Reader, int64) (interface{}, error)

const Nil = RedisError("redis: nil") // nolint:errname

type RedisError string

func (e RedisError) Error() string { return string(e) }

func (RedisError) RedisError() {}

type Reader struct {
	rd   *bufio.Reader
	_buf []byte
}

基础字节流

func (r *Reader) Buffered() int，缓冲区内容大小
func (r *Reader) Peek(n int) ([]byte, error)，读取但不偏移
func (r *Reader) Reset(rd io.Reader)，重置
func (r *Reader) ReadLine() ([]byte, error)，读取下一行

解析协议

func (r *Reader) ReadReply(m MultiBulkParse) (interface{}, error)，解析响应，m为解析多行的函数
func (r *Reader) ReadIntReply() (int64, error)，解析数字
func (r *Reader) ReadString() (string, error)，解析字符串
func (r *Reader) ReadArrayReply(m MultiBulkParse) (interface{}, error)，解析数组
func (r *Reader) ReadArrayLen() (int, error)，解析数组长度
func (r *Reader) ReadScanReply() ([]string, uint64, error)，解析scan数据

读取下一行内容并解析成指定类型

func (r *Reader) readTmpBytesReply() ([]byte, error)，读取一行字节流内容
func (r *Reader) ReadInt() (int64, error)，将字节流解析有符号数字
func (r *Reader) ReadUint() (uint64, error)，将字节流解析无符号数字
func (r *Reader) ReadFloatReply() (float64, error)，将字节流解析浮点

scan

解析结构体的redis tag，利用反射设置

func Scan(b []byte, v interface{}) error，将字节流的值反射入v
func ScanSlice(data []string, slice interface{}) error，将字符串数组反射入slice

net

Conn

封装对resp协议的读写

type Conn struct {
	usedAt  int64 // atomic
	netConn net.Conn

	rd *proto.Reader
	bw *bufio.Writer // 用于写字节
	wr *proto.Writer // 用于写协议

	Inited    bool // 是否初始化
	pooled    bool // 是否归还池
	createdAt time.Time
}

连接属性

func (cn *Conn) UsedAt() time.Time，获取上次使用时间
func (cn *Conn) SetUsedAt(tm time.Time)，设置使用时间
func (cn *Conn) SetNetConn(netConn net.Conn)，设置底层网络连接
func (cn *Conn) RemoteAddr() net.Addr，获取远端地址

连接操作

func (cn *Conn) Write(b []byte) (int, error)，写入二进制数据
func (cn *Conn) WithReader(ctx context.Context, timeout time.Duration, fn func(rd *proto.Reader) error) error，带有reader的读协议
func (cn *Conn) WithWrite(ctx context.Context, timeout time.Duration, fn func(wr *proto.Writer) error，带有writer的写协议
func (cn *Conn) Close() error，关闭连接

Pooler

连接池接口，用于管控连接池连接

type Pooler interface {
	NewConn(context.Context) (*Conn, error)
	CloseConn(*Conn) error

	Get(context.Context) (*Conn, error)
	Put(context.Context, *Conn)
	Remove(context.Context, *Conn, error)

	Len() int
	IdleLen() int
	Stats() *Stats

	Close() error
}

ConnPool

基础连接池

type ConnPool struct {
	opt *Options

	dialErrorsNum uint32 // atomic

	lastDialError atomic.Value

	queue chan struct{} // 并发控制

	connsMu      sync.Mutex
	conns        []*Conn // 所有连接的连接对象 包含不是池连接
	idleConns    []*Conn // 空闲连接
	poolSize     int     // 分配出去的池连接个数
	idleConnsLen int

	stats Stats

	_closed  uint32 // atomic
	closedCh chan struct{}
}

type Options struct {
	Dialer  func(context.Context) (net.Conn, error) // 拨号器
	OnClose func(*Conn) error                       // 关闭事件处理

	PoolFIFO           bool // 是否是先进先出
	PoolSize           int
	MinIdleConns       int
	MaxConnAge         time.Duration
	PoolTimeout        time.Duration
	IdleTimeout        time.Duration
	IdleCheckFrequency time.Duration
}

type Stats struct {
	Hits     uint32 // number of times free connection was found in the pool
	Misses   uint32 // number of times free connection was NOT found in the pool
	Timeouts uint32 // number of times a wait timeout occurred

	TotalConns uint32 // number of total connections in the pool
	IdleConns  uint32 // number of idle connections in the pool
	StaleConns uint32 // number of stale connections removed from the pool
}

并发的控制

func (p *ConnPool) getTurn()，阻塞获取执行权限
func (p *ConnPool) waitTurn(ctx context.Context) error，等待获取执行权限
func (p *ConnPool) freeTurn()，释放执行权限

池化的管理

func (p *ConnPool) Get(ctx context.Context) (*Conn, error)，获取连接
func (p *ConnPool) Put(ctx context.Context, cn *Conn)，归还连接
func (p *ConnPool) Remove(ctx context.Context, cn *Conn, reason error)，移除连接
func (p *ConnPool) CloseConn(cn *Conn) error，关闭连接
func (p *ConnPool) Len() int，连接池连接数
func (p *ConnPool) IdleLen() int，空闲连接数
func (p *ConnPool) Stats() *Stats，返回状态
func (p *ConnPool) Close() error，关闭连接池
func (p *ConnPool) Filter(fn func(*Conn) bool) error，对连接池每个连接执行过滤函数fn

异步清理空闲连接

func (p *ConnPool) reaper(frequency time.Duration)，定时清理过期连接
func (p *ConnPool) ReapStaleConns() (int, error)，清理过期连接
func (p *ConnPool) isStaleConn(cn *Conn) bool，判断是否是过期连接

SingleConnPool

池里面只有一个连接的连接池用于有连接状态的请求连接池

type SingleConnPool struct {
	pool      Pooler
	cn        *Conn
	stickyErr error
}

复写底层pool的方法

func (p *SingleConnPool) NewConn(ctx context.Context) (*Conn, error)，新建连接
func (p *SingleConnPool) CloseConn(cn *Conn) error，关闭连接
func (p *SingleConnPool) Get(ctx context.Context) (*Conn, error)，获取连接
func (p *SingleConnPool) Put(ctx context.Context, cn *Conn) ，归还连接
func (p *SingleConnPool) Remove(ctx context.Context, cn *Conn, reason error)，移除连接
func (p *SingleConnPool) Close() error，关闭连接池

StickyConnPool

黏性连接池，用于事务连接管理，可主动关闭连接

type StickyConnPool struct {
	pool   Pooler
	// 引用计数
	shared int32 // atomic

	state uint32 // atomic
	// 单个连接的通道数据
	ch    chan *Conn

	_badConnError atomic.Value
}

复写底层pool方法

func (p *StickyConnPool) NewConn(ctx context.Context) (*Conn, error)，新建连接
func (p *StickyConnPool) CloseConn(cn *Conn) error，关闭连接
func (p *StickyConnPool) Get(ctx context.Context) (*Conn, error)，获取连接
func (p *StickyConnPool) Put(ctx context.Context, cn *Conn)，归还连接
func (p *StickyConnPool) Remove(ctx context.Context, cn *Conn, reason error)，移除连接
func (p *StickyConnPool) Close() error，关闭连接池
func (p *StickyConnPool) Len() int，连接池连接数
func (p *StickyConnPool) IdleLen() int，空闲连接数
func (p *StickyConnPool) Stats() *Stats，连接池状态

重置操作

func (p *StickyConnPool) Reset(ctx context.Context) error，重置连接

总结

命令操作使用函数式编程，让调用更紧密，上层不需要关心底层逻辑
各模块之间使用接口，让底层实现更加灵活
三层结构（操作-连接-协议）层次分明，各司其职，结构清晰

djaigo

https://blog.djaigo.com/golang/go-redis.html

本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 djaigo !

redis

Golang Map 内部实现详解

2022-04-14 golang

golang map 数据结构源码分析

golang-单元测试

2021-05-13 golang

go-redis

go-redis

call chain

core object

client

option

hook

baseClient

Client

Conn

Tx

command

commands

command

SliceCmd

StatusCmd

IntCmd

IntSliceCmd

DurationCmd

TimeCmd

BoolCmd

StringCmd

FloatCmd

FloatSliceCmd

StringSliceCmd

BoolSliceCmd

StringStringMapCmd

StringIntMapCmd

StringStructMapCmd

ZSliceCmd

ZWithKeyCmd

ScanCmd

proto

request

reply

scan

net

Conn

Pooler

ConnPool

SingleConnPool

StickyConnPool

总结

程序猿唯一赚钱方式!<br/>The only way to make money for code monkey!