2. 程式人生 > >Go語言之從0到1實現一個簡單的Redis連線池

Go語言之從0到1實現一個簡單的Redis連線池

阿新 • • 發佈：2019-07-03

Go語言之從0到1實現一個簡單的Redis連線池

前言

最近學習了一些Go語言開發相關內容，但是苦於手頭沒有可以練手的專案，學的時候理解不清楚，學過容易忘。

結合之前組內分享時學到的Redis相關知識，以及Redis Protocol文件，就想著自己造個輪子練練手。

這次我把目標放在了Redis client implemented with Go，使用原生Go語言和TCP實現一個簡單的Redis連線池和協議解析，以此來讓自己入門Go語言，並加深理解和記憶。（這樣做直接導致的後果是，最近寫JS時if語句總是忘帶括號QAQ）。

本文只能算是學習Go語言時的一個隨筆，並不是真正要造一個線上環境可用的Go-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結尾，在命令結構的開頭就聲明瞭命令的引數數量，每一條引數都帶有長度標記，方便服務端解析。

例如，傳送一個SET命令set name jeferwang：

*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

實現流程

  1. 首先，我們根據官網文件瞭解到了Redis傳輸協議，即Redis使用TCP傳輸命令的格式和接收資料的格式，據此，我們可以使用Go實現對Redis協議的解析
  2. 接下來，在可以建立Redis連線並進行資料傳輸的前提下，實現一個連線池。
  3. 實現拼接Redis命令的方法，通過TCP傳送到RedisServer
  4. 讀取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--

相關推薦

Go語言之從0到1實現一個簡單的Redis連線池

Go語言之從0到1實現一個簡單的Redis連線池 前言 最近學習了一些Go語言開發相關內容，但是苦於手頭沒有可以練手的專案，學的時候理解不清楚，學過容易忘。 結合之前組內分享時學到的Redis相關知識，以及Redis Protocol文件，就想著自己造個輪子練練手。 這次我把目標放在了Redis client

【go語言 socket程式設計系列】一個簡單的HTTP伺服器及func (srv *Server) Serve(l net.Listener) 方法

【簡單的HTTP伺服器】 原始檔server.go中 ListenAndServe()函式的註釋中有個簡單的HTTP服務實現程式碼，如下 package main import ( "io" "log" "net/http" ) func HelloServ

用go的goroutine和channel實現一個簡單的“生產、消費”（帶有超時控制）小例子

直接上程式碼 1、沒有超時的時候 package main import ( "fmt" "time" ) func main() { //存放生產的ch

一個簡單資料庫連線池的實現

一、已實現功能   資料庫連線快取。將資料庫連線與執行緒ID繫結並提供執行資料庫操作時檢測。資料庫連線超時檢測。初始化資料庫環境，包括初始化資料庫，資料庫使用者，資料庫表。 二、程式碼列表： 1、MySqlDBManager：   用於管理資料庫配置、初始化資料庫環境及建立

go 語言實現一個簡單的 web 伺服器

學習Go語言的一些感受,不一定準確。假如發生戰爭，JAVA一般都是充當航母戰鬥群的角色。一旦出動,就是護衛艦、巡洋艦、航母艦載機、預警機、電子戰飛機、潛艇等等浩浩蕩蕩,殺將過去。(JVM,數十個JAR包,Tomcat中介軟體,SSH框架,各種配置檔案...天生就是重量級的,

Go語言入門——從一個簡單例子入門

檢視文件 首先先分享一個可以在本地就能檢視文件的騷操作（linux系統） 1.開啟命令列終端，輸入godoc -http=:8000,如果想後臺執行在後面加個& 2.然後開啟瀏覽器，輸入網址：localhost:8000 就可以愉快地看文件了 go語言是個什麼東西

從零開始打造一個新聞訂閱APP之爬蟲篇（二、實現一個簡單的爬蟲系統）

前景提要：如何開發一個新聞訂閱APP之爬蟲篇（一、背景介紹&需求分析） 做一個特定的爬蟲系統，首先考慮它要做什麼？ 從網際網路上抓取指定的N個站點資訊，解析提取需要的內容，按照特定的結構儲存； 系統結構圖如下： 下面是主要的程式碼結構；

GO語言實現一個簡單的網頁爬蟲

具體思路是利用go語言提供的遠端包github.com/PuerkitoBio/goquery，但是有時候連結會出錯，導致程式編譯不通過，我們可以直接將這個包下載到本地，然後將import匯入地址修改為本地的goquery地址即可首先建立檔案，儲存從網頁上扒下來的內容接下來使

05.從0實現一個JVM語言之目標平臺程式碼生成-CodeGenerator

從0實現JVM語言之目標平臺程式碼生成-CodeGenerator === ## [原始碼github倉庫, 如果這個系列文章對你有幫助, 希望獲得你的一個star](https://github.com/MilitaryIntelligence6/Cva) ## [本節相關程式碼生成package地

JavaScript之實現一個簡單的Vue

方法 ole tro def pre 一個 進行 this upd vue的使用相信大家都很熟練了，使用起來簡單。但是大部分人不知道其內部的原理是怎麽樣的，今天我們就來一起實現一個簡單的vue。 Object.defineProperty() 實現之前我們得先看一下Obje

java學習之—實現一個簡單的ArrayList

package thread1; /** * 實現一個簡單的ArrayList * * @Title: uminton */ public class SimpleArrayList<T> { //陣列中元素的大小 private Intege

用c語言實現一個簡單的通訊錄

通訊錄的c語言實現原始碼 簡單通訊錄的實現還是包括三個原始檔，test.c（實現通訊錄主邏輯），txl.c（實現用到的各個函式），txl.h（存放txl中用到的各種標頭檔案與宣告）。 txl.h #ifndef __TXL_H__//**txl.h** #defi

JavaScript 之實現一個簡單的 Vue

vue的使用相信大家都很熟練了，使用起來簡單。但是大部分人不知道其內部的原理是怎麼樣的，今天我們就來一起實現一個簡單的vue Object.defineProperty() 實現之前我們得先看一下Object.defineProperty的實現，因為vue主要是通過資料劫

LayaBox開發實戰之實現一個簡單的模板類

tom image nbu tex 技術分享 info fig utils gre 1.首先UI設計：記得導出 2.然後查看layaUI.max.all.js中是否生成對應UI的JS代碼： var FeedBackUI=(function(_super){

.NET Core微服務之路：利用DotNetty實現一個簡單的通訊過程

　　上一篇我們已經全面的介紹過《基於gRPC服務發現與服務治理的方案》，我們先複習一下RPC的呼叫過程（筆者會在這一節的幾篇文章中反覆的強調這個過程呼叫方案），看下圖

實現一個簡單的Web伺服器（C語言）

Web伺服器 github地址 該專案的第二部分是在第一部分的基礎上繼續完善Web伺服器。 第二部分主要是完成兩個功能：記錄日誌和HTTP響應。 記錄日誌 該部分比較簡單，只要簡單地將一些伺服器的資訊新增到指定日誌檔案即可， 也不需要完成很複雜的功能，有了日誌功

JAVA實現一個簡單的代數運算語言編譯器（二）--詞法分析準備

上一篇文章主要介紹了這個代數運算編譯器的起因，這一篇我們就來開始寫這個專案。 首先我們需要先設定一些系統的基礎類如系統符號類，保留字類、錯誤提示資訊類、自定義異常、輸入讀取類等，下面簡單地說一下這幾個類。 系統符號類： package com.liu.system; /

go實現一個簡單的遊戲伺服器框架（lotou）基本設計

程式碼倉庫，目前程式碼比較粗糙，歡迎各種改進建議。 因為go語言原生支援高併發（goroutine）和通訊機制（channel），所以首先想到的就是使用這兩個東西來構建我的訊息分發機制。 核心思路： 有一個core模組維護了所有的服務 當有訊息傳送的時候

go實現一個簡單的遊戲伺服器框架（lotou）起源

程式碼倉庫，目前程式碼比較粗糙，歡迎各種改進建議。 最近一直想學習一些關於遊戲伺服器的知識，顯示看了一下雲風的skynet框架，從而對於一個遊戲伺服器框架有了一個基本概要了解。先來說說我對於skynet的一些理解吧。 skynet理解 skynet只完成了

go實現一個簡單的遊戲伺服器框架（lotou）編碼

程式碼倉庫 在lotou中實現了兩種二進位制編碼方式。 binary 一種是普通的C/C++模式的小端編碼方式，每一個數據都被依次編碼進一個二進位制流中，基本用法如下： 編碼 binary.Encode(v) binary.Encode(v)