Go-Redigo-源码分析一-实现Protocol协议请求redis

概述

Redis 是我们日常开发中使用的最常见的一种 Nosql, 是一个 key-value 存储系统，但是 redis 不止支持 key-value，还自持很多存储类型包括字符串、链表、集合、有序集合和哈希。
在 go 使用 redis 中有很多的开源库可以使用，我经常使用的是 redigo 这个库，它封装很多对 redis 的 api、网络链接和连接池。
分析 Redigo 之前我觉得需要知道如果不用 redigo，我们该如何访问 redis。之后才能更加简单方便的理解 Redigo 是做了一些什么事。

Protocol 协议

官方对 protocol 协议的定义：链接

网络层：

客户端和服务端用通过 TCP 链接来交互

请求

*< 参数数量 > CR LF
$< 参数 1 的字节数量 > CR LF
< 参数 1 的数据 > CR LF
…
$< 参数 N 的字节数量 > CR LF
< 参数 N 的数据 > CR LF
举个例子 get aaa = *2rn$3\r\nget\r\n$3rn$aaarn
每个参数结尾用 rn $ 之后是参数的字节数
这样组成的一串命令通过 tcp 发送到 redis 服务端之后就是 redis 的返回了

返回

Redis 的返回有 5 中情况：

• 状态回复（status reply）的第一个字节是 “+”
• 错误回复（error reply）的第一个字节是 “-“
• 整数回复（integer reply）的第一个字节是 “:”
• 批量回复（bulk reply）的第一个字节是 “$”
• 多条批量回复（multi bulk reply）的第一个字节是 “*”

下面按照 5 中情况各自举一个例子

状态回复 ：
请求: set aaa aaa
回复: +OKrn
错误回复 ：
请求: set aaa
回复: -ERR wrong number of arguments for ‘set’ commandrn
整数回复 ：
请求：llen list
回复：:5rn
批量回复
请求: get aaa
回复: $3rnaaarn
多条批量回复
请求: lrange list 0 -1
回复: *3rn$3\r\naaa\r\n$3rndddrn$3rncccrn

实现

那么我们如何用 go 来实现不用 redis 框架，自己请求 redis 服务。其实也很简单，go 提供很方便的 net 包让我们很容易的使用 tcp
先看解析回复方法，封装了一个 reply 对象：

package client

import (
    "bufio"
    "errors"
    "fmt"
    "net"
    "strconv"
)

type Reply struct {
    Conn        *net.TCPConn
    SingleReply []byte
    MultiReply  [][]byte
    Source      []byte
    IsMulti     bool
    Err         error
}

// 组成请求命令
func MultiCommandMarshal(args ...string) string {
    var s string
    s = "*"
    s += strconv.Itoa(len(args))
    s += "\r\n"

    // 命令所有参数
    for _, v := range args {
        s += "$"
        s += strconv.Itoa(len(v))
        s += "\r\n"
        s += v
        s += "\r\n"
    }

    return s
}

// 预读取第一个字节判断是多行还是单行返回 分开处理
func (reply *Reply) Reply() {rd := bufio.NewReader(reply.Conn)
    b, err := rd.Peek(1)

    if err != nil {fmt.Println("conn error")
    }
    fmt.Println("prefix =", string(b))
    if b[0] == byte('*') {
        reply.IsMulti = true
        reply.MultiReply, reply.Err = multiResponse(rd)
    } else {
        reply.IsMulti = false
        reply.SingleReply, err = singleResponse(rd)
        if err != nil {
            reply.Err = err
            return
        }
    }
}

// 多行返回 每次读取一行然后调用 singleResponse 获取单行数据
func multiResponse(rd *bufio.Reader) ([][]byte, error) {prefix, err := rd.ReadByte()
    var result [][]byte
    if err != nil {return result, err}
    if prefix != byte('*') {return result, errors.New("not multi response")
    }
    //*3\r\n$1\r\n3\r\n$1\r\n2\r\n$1\r\n
    l, _, err := rd.ReadLine()
    if err != nil {return result, err}
    n, err := strconv.Atoi(string(l))
    if err != nil {return result, err}
    for i := 0; i < n; i++ {s, err := singleResponse(rd)
        fmt.Println("i =", i, "result =", string(s))
        if err != nil {return result, err}
        result = append(result, s)
    }

    return result, nil
}

// 获取单行数据 + -：逻辑相同 $ 单独处理
func singleResponse(rd *bufio.Reader) ([]byte, error) {
    var (result []byte
        err    error
    )
    prefix, err := rd.ReadByte()
    if err != nil {return []byte{}, err}
    switch prefix {case byte('+'), byte('-'), byte(':'):
        result, _, err = rd.ReadLine()
    case byte('$'):
        // $7\r\nliangwt\r\n
        n, _, err := rd.ReadLine()
        if err != nil {return []byte{}, err}
        l, err := strconv.Atoi(string(n))
        if err != nil {return []byte{}, err}
        p := make([]byte, l+2)
        rd.Read(p)
        result = p[0 : len(p)-2]

    }

    return result, err
}

然后看下如何调用

package main

import (
    "bufio"
    "flag"
    "fmt"
    "log"
    "net"
    "os"
    "strconv"
    "strings"
    "test/redis/rediscli/client"
)

var host string
var port string

func init() {
    // 参数获取 设置有默认值
    flag.StringVar(&host, "h", "localhost", "hsot")
    flag.StringVar(&port, "p", "6379", "port")
}

func main() {flag.Parse()

    porti, err := strconv.Atoi(port)
    if err != nil {panic("port is error")
    }
    
    tcpAddr := &net.TCPAddr{IP: net.ParseIP(host), Port: porti}
    conn, err := net.DialTCP("tcp", nil, tcpAddr)
    if err != nil {log.Println(err)
    }
    defer conn.Close()

    for {fmt.Printf("%s:%d>", host, porti)
        bio := bufio.NewReader(os.Stdin)
        input, _, err := bio.ReadLine()
        if err != nil {fmt.Println(err)
        }
        s := strings.Split(string(input), " ")
        req := client.MultiCommandMarshal(s...)
        conn.Write([]byte(req))
        reply := client.Reply{}
        reply.Conn = conn
        reply.Reply()

        if reply.Err != nil {fmt.Println("err:", reply.Err)
        }
        var res []byte
        if reply.IsMulti { } else {res = reply.SingleReply}
        fmt.Println("result:", string(res), "\nerr:", err)
        //fmt.Println(string(p))
    }

}

总结

上面的代码我们看到根据不同的回复类型，用不同的逻辑解析。
其实所有的 redis 处理框架的本质就是封装上面的代码，让我们使用更加方便。当然还有一些其他的功能 使用 Lua 脚本、发布订阅等等功能。
我觉得要理解 redis 库 首先要理解 Protocol，然后再去看源码 否则你会看到很多你看不懂的逻辑和封装。所以先研究了下 Protocol 协议并自己实现了一下。