Go语言之从0到1实现一个简单的Redis链接池

时间 2019-11-19

标签语言实现一个简单 redis 链接栏目 Redis 繁體版

原文原文链接

Go语言之从0到1实现一个简单的Redis链接池

前言

最近学习了一些Go语言开发相关内容，可是苦于手头没有能够练手的项目，学的时候理解不清楚，学过容易忘。html

结合以前组内分享时学到的Redis相关知识，以及Redis Protocol文档，就想着本身造个轮子练练手。git

此次我把目标放在了Redis client implemented with Go，使用原生Go语言和TCP实现一个简单的Redis链接池和协议解析，以此来让本身入门Go语言，并加深理解和记忆。（这样作直接致使的后果是，最近写JS时if语句老是忘带括号QAQ）。github

本文只能算是学习Go语言时的一个随笔，并非真正要造一个线上环境可用的Go-Redis库~(︿(￣︶￣)︿摊手)redis

顺便安利如下本身作的一个跨平台开源Redis管理软件：AwesomeRedisManager官网 AwesomeRedisManager源码数据库

Redis协议主要参考这篇文档通讯协议（protocol），阅读后了解到，Redis Protocol并无什么复杂之处，主要是使用TCP来传输一些固定格式的字符串数据达到发送命令和解析Response数据的目的。数组

命令格式

根据文档了解到，Redis命令格式为（CR LF即\r\n）：服务器

*<参数数量N> CR LF
$<参数 1 的字节数量> CR LF
<参数 1 的数据> CR LF
...
$<参数 N 的字节数量> CR LF
<参数 N 的数据> CR LF

命令的每一行都使用CRLF结尾，在命令结构的开头就声明了命令的参数数量，每一条参数都带有长度标记，方便服务端解析。app

例如，发送一个SET命令set name jeferwang：tcp

*3
$3
SET
$4
name
$9
jeferwang

响应格式

Redis的响应回复数据主要分为五种类型：性能

状态回复：一行数据，使用+开头（例如：OK、PONG等）

+OK\r\n
+PONG\r\n

错误回复：一行数据，使用-开头（Redis执行命令时产生的错误）

-ERR unknown command 'demo'\r\n

整数回复：一行数据，使用:开头（例如：llen返回的长度数值等）

:100\r\n

批量回复（能够理解为字符串）：两行数据，使用$开头，第一行为内容长度，第二行为具体内容

$5\r\n
abcde\r\n

特殊状况：$-1\r\n即为返回空数据，能够转化为nil

多条批量回复：使用*开头，第一行标识本次回复包含多少条批量回复，后面每两行为一个批量回复（lrange、hgetall等命令的返回数据）

*2\r\n
$5\r\n
ABCDE\r\n
$2\r\n
FG\r\n

更详细的命令和回复格式能够从Redis Protocol文档了解到，本位只介绍一些基本的开发中须要用到的内容

如下为部分代码，完整代码见GitHub：redis4go

实现流程

首先，咱们根据官网文档了解到了Redis传输协议，即Redis使用TCP传输命令的格式和接收数据的格式，据此，咱们可使用Go实现对Redis协议的解析
接下来，在能够创建Redis链接并进行数据传输的前提下，实现一个链接池。
实现拼接Redis命令的方法，经过TCP发送到RedisServer
读取RedisResponse，实现解析数据的方法

模块结构分析

简单分析Redis链接池的结构，能够先简单规划为5个部分：

结构体定义entity.go
Redis链接和调用redis_conn.go
Redis数据类型解析data_type.go
链接池实现pool.go

共划分为上述四个部分

对象结构定义

为了实现链接池及Redis数据库链接，咱们须要以下结构：

Redis服务器配置RedisConfig：包含Host、Port等信息
Redis链接池配置PoolConfig：继承RedisConfig，包含PoolSize等信息
Redis链接池结构：包含链接队列、链接池配置等信息
单个Redis链接：包含TCP链接Handler、是否处于空闲标记位、当前使用的数据库等信息

package redis4go

import (
    "net"
    "sync"
)

type RedisConfig struct {
    Host     string // RedisServer主机地址
    Port     int    // RedisServer主机端口
    Password string // RedisServer须要的Auth验证，不填则为空
}

// 链接池的配置数据
type PoolConfig struct {
    RedisConfig
    PoolSize int // 链接池的大小
}

// 链接池结构
type Pool struct {
    Config PoolConfig          // 创建链接池时的配置
    Queue  chan *RedisConn     // 链接池
    Store  map[*RedisConn]bool // 全部的链接
    mu     sync.Mutex          // 加锁
}

// 单个Redis链接的结构
type RedisConn struct {
    mu        sync.Mutex   // 加锁
    p         *Pool        // 所属的链接池
    IsRelease bool         // 是否处于释放状态
    IsClose   bool         // 是否已关闭
    TcpConn   *net.TCPConn // 创建起的到RedisServer的链接
    DBIndex   int          // 当前链接正在使用第几个Redis数据库
}

type RedisResp struct {
    rType byte     // 回复类型(+-:$*)
    rData [][]byte // 从TCP链接中读取的数据统一使用二维数组返回
}

根据以前的规划，定义好基本的结构以后，咱们能够先实现一个简单的Pool对象池

Redis链接

创建链接

首先咱们须要实现一个创建Redis链接的方法

// 建立一个RedisConn对象
func createRedisConn(config RedisConfig) (*RedisConn, error) {
    tcpAddr := &net.TCPAddr{IP: net.ParseIP(config.Host), Port: config.Port}
    tcpConn, err := net.DialTCP("tcp", nil, tcpAddr)
    if err != nil {
        return nil, err
    }
    return &RedisConn{
        IsRelease: true,
        IsClose:   false,
        TcpConn:   tcpConn,
        DBIndex:   0,
    }, nil
}

实现链接池

在Go语言中，咱们可使用一个chan来很轻易地实现一个指定容量的队列，来做为链接池使用，当池中没有链接时，申请获取链接时将会被阻塞，直到放入新的链接。

package redis4go

func CreatePool(config PoolConfig) (*Pool, error) {
    pool := &Pool{
        Config: config,
        Queue:  make(chan *RedisConn, config.PoolSize),
        Store:  make(map[*RedisConn]bool, config.PoolSize),
    }
    for i := 0; i < config.PoolSize; i++ {
        redisConn, err := createRedisConn(config.RedisConfig)
        if err != nil {
            // todo 处理以前已经建立好的连接
            return nil, err
        }
        redisConn.p = pool
        pool.Queue <- redisConn
        pool.Store[redisConn] = true
    }
    return pool, nil
}

// 获取一个链接
func (pool *Pool) getConn() *RedisConn {
    pool.mu.Lock()
    // todo 超时机制
    conn := <-pool.Queue
    conn.IsRelease = false
    pool.mu.Unlock()
    return conn
}

// 关闭链接池
func (pool *Pool) Close() {
    for conn := range pool.Store {
        err := conn.Close()
        if err != nil {
            // todo 处理链接关闭的错误？
        }
    }
}

发送命令&解析回复数据

下面是向RedisServer发送命令，以及读取回复数据的简单实现

func (conn *RedisConn) Call(params ...interface{}) (*RedisResp, error) {
    reqData, err := mergeParams(params...)
    if err != nil {
        return nil, err
    }
    conn.Lock()
    defer conn.Unlock()
    _, err = conn.TcpConn.Write(reqData)
    if err != nil {
        return nil, err
    }
    resp, err := conn.getReply()
    if err != nil {
        return nil, err
    }
    if resp.rType == '-' {
        return resp, resp.ParseError()
    }
    return resp, nil
}

func (conn *RedisConn) getReply() (*RedisResp, error) {
    b := make([]byte, 1)
    _, err := conn.TcpConn.Read(b)
    if err != nil {
        return nil, err
    }
    resp := new(RedisResp)
    resp.rType = b[0]
    switch b[0] {
    case '+':
        // 状态回复
        fallthrough
    case '-':
        // 错误回复
        fallthrough
    case ':':
        // 整数回复
        singleResp := make([]byte, 1)
        for {
            _, err := conn.TcpConn.Read(b)
            if err != nil {
                return nil, err
            }
            if b[0] != '\r' && b[0] != '\n' {
                singleResp = append(singleResp, b[0])
            }
            if b[0] == '\n' {
                break
            }
        }
        resp.rData = append(resp.rData, singleResp)
    case '$':
        buck, err := conn.readBuck()
        if err != nil {
            return nil, err
        }
        resp.rData = append(resp.rData, buck)
    case '*':
        // 条目数量
        itemNum := 0
        for {
            _, err := conn.TcpConn.Read(b)
            if err != nil {
                return nil, err
            }
            if b[0] == '\r' {
                continue
            }
            if b[0] == '\n' {
                break
            }
            itemNum = itemNum*10 + int(b[0]-'0')
        }
        for i := 0; i < itemNum; i++ {
            buck, err := conn.readBuck()
            if err != nil {
                return nil, err
            }
            resp.rData = append(resp.rData, buck)
        }
    default:
        return nil, errors.New("错误的服务器回复")
    }
    return resp, nil
}

func (conn *RedisConn) readBuck() ([]byte, error) {
    b := make([]byte, 1)
    dataLen := 0
    for {
        _, err := conn.TcpConn.Read(b)
        if err != nil {
            return nil, err
        }
        if b[0] == '$' {
            continue
        }
        if b[0] == '\r' {
            break
        }
        dataLen = dataLen*10 + int(b[0]-'0')
    }
    bf := bytes.Buffer{}
    for i := 0; i < dataLen+3; i++ {
        _, err := conn.TcpConn.Read(b)
        if err != nil {
            return nil, err
        }
        bf.Write(b)
    }
    return bf.Bytes()[1 : bf.Len()-2], nil
}

func mergeParams(params ...interface{}) ([]byte, error) {
    count := len(params) // 参数数量
    bf := bytes.Buffer{}
    // 参数数量
    {
        bf.WriteString("*")
        bf.WriteString(strconv.Itoa(count))
        bf.Write([]byte{'\r', '\n'})
    }
    for _, p := range params {
        bf.Write([]byte{'$'})
        switch p.(type) {
        case string:
            str := p.(string)
            bf.WriteString(strconv.Itoa(len(str)))
            bf.Write([]byte{'\r', '\n'})
            bf.WriteString(str)
            break
        case int:
            str := strconv.Itoa(p.(int))
            bf.WriteString(strconv.Itoa(len(str)))
            bf.Write([]byte{'\r', '\n'})
            bf.WriteString(str)
            break
        case nil:
            bf.WriteString("-1")
            break
        default:
            // 不支持的参数类型
            return nil, errors.New("参数只能是String或Int")
        }
        bf.Write([]byte{'\r', '\n'})
    }
    return bf.Bytes(), nil
}

实现几个经常使用数据类型的解析

package redis4go

import (
    "errors"
    "strconv"
)

func (resp *RedisResp) ParseError() error {
    if resp.rType != '-' {
        return nil
    }
    return errors.New(string(resp.rData[0]))
}

func (resp *RedisResp) ParseInt() (int, error) {
    switch resp.rType {
    case '-':
        return 0, resp.ParseError()
    case '$':
        fallthrough
    case ':':
        str, err := resp.ParseString()
        if err != nil {
            return 0, err
        }
        return strconv.Atoi(str)
    default:
        return 0, errors.New("错误的回复类型")
    }
}

func (resp *RedisResp) ParseString() (string, error) {
    switch resp.rType {
    case '-':
        return "", resp.ParseError()
    case '+':
        fallthrough
    case ':':
        fallthrough
    case '$':
        return string(resp.rData[0]), nil
    default:
        return "", errors.New("错误的回复类型")
    }
}
func (resp *RedisResp) ParseList() ([]string, error) {
    switch resp.rType {
    case '-':
        return nil, resp.ParseError()
    case '*':
        list := make([]string, 0, len(resp.rData))
        for _, data := range resp.rData {
            list = append(list, string(data))
        }
        return list, nil
    default:
        return nil, errors.New("错误的回复类型")
    }
}
func (resp *RedisResp) ParseMap() (map[string]string, error) {
    switch resp.rType {
    case '-':
        return nil, resp.ParseError()
    case '*':
        mp := make(map[string]string)
        for i := 0; i < len(resp.rData); i += 2 {
            mp[string(resp.rData[i])] = string(resp.rData[i+1])
        }
        return mp, nil
    default:
        return nil, errors.New("错误的回复类型")
    }
}

在开发的过程当中，随手编写了几个零零散散的测试文件，经测试，一些简单的Redis命令以及能跑通了。

package redis4go

import (
    "testing"
)

func getConn() (*RedisConn, error) {
    pool, err := CreatePool(PoolConfig{
        RedisConfig: RedisConfig{
            Host: "127.0.0.1",
            Port: 6379,
        },
        PoolSize: 10,
    })
    if err != nil {
        return nil, err
    }
    conn := pool.getConn()
    return conn, nil
}

func TestRedisResp_ParseString(t *testing.T) {
    demoStr := string([]byte{'A', '\n', '\r', '\n', 'b', '1'})
    conn, _ := getConn()
    _, _ = conn.Call("del", "name")
    _, _ = conn.Call("set", "name", demoStr)
    resp, err := conn.Call("get", "name")
    if err != nil {
        t.Fatal("Call Error:", err.Error())
    }
    str, err := resp.ParseString()
    if err != nil {
        t.Fatal("Parse Error:", err.Error())
    }
    if str != demoStr {
        t.Fatal("结果错误")
    }
}

func TestRedisResp_ParseList(t *testing.T) {
    conn, _ := getConn()
    _, _ = conn.Call("del", "testList")
    _, _ = conn.Call("lpush", "testList", 1, 2, 3, 4, 5)
    res, err := conn.Call("lrange", "testList", 0, -1)
    if err != nil {
        t.Fatal("Call Error:", err.Error())
    }
    ls, err := res.ParseList()
    if err != nil {
        t.Fatal("Parse Error:", err.Error())
    }
    if len(ls) != 5 {
        t.Fatal("结果错误")
    }
}

func TestRedisResp_ParseMap(t *testing.T) {
    conn, _ := getConn()
    _, _ = conn.Call("del", "testMap")
    _, err := conn.Call("hmset", "testMap", 1, 2, 3, 4, 5, 6)
    if err != nil {
        t.Fatal("设置Value失败")
    }
    res, err := conn.Call("hgetall", "testMap")
    if err != nil {
        t.Fatal("Call Error:", err.Error())
    }
    ls, err := res.ParseMap()
    if err != nil {
        t.Fatal("Parse Error:", err.Error())
    }
    if len(ls) != 3 || ls["1"] != "2" {
        t.Fatal("结果错误")
    }
}

至此，已经算是达到了学习Go语言和学习Redis Protocol的目的，不过代码中也有不少地方须要优化和完善，性能方面考虑的也并不周全。轮子就不重复造了，毕竟有不少功能完善的库，从头造一个轮子须要消耗的精力太多啦而且不必~

下一次我将会学习官方推荐的gomodule/redigo源码，并分享个人心得。

--The End--