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協議並本身實現了一下。