引言

用11篇文章實(shí)現(xiàn)一個(gè)可用的Redis服務(wù)，姑且叫EasyRedis吧，希望通過(guò)文章將Redis掰開撕碎了呈現(xiàn)給大家，而不是僅僅停留在八股文的層面，并且有非常爽的感覺(jué)，歡迎持續(xù)關(guān)注學(xué)習(xí)。

[x] easyredis之TCP服務(wù)

[x] easyredis之網(wǎng)絡(luò)請(qǐng)求序列化協(xié)議（RESP）

[ ] easyredis之內(nèi)存數(shù)據(jù)庫(kù)

[ ] easyredis之過(guò)期時(shí)間 (時(shí)間輪實(shí)現(xiàn))

[ ] easyredis之持久化 （AOF實(shí)現(xiàn)）

[ ] easyredis之發(fā)布訂閱功能

[ ] easyredis之有序集合（跳表實(shí)現(xiàn)）

[ ] easyredis之 pipeline 客戶端實(shí)現(xiàn)

[ ] easyredis之事務(wù)（原子性/回滾）

[ ] easyredis之連接池

[ ] easyredis之分布式集群存儲(chǔ)

EasyRedis之網(wǎng)絡(luò)請(qǐng)求序列化協(xié)議（RESP）

Redis 協(xié)議格式

全名叫: Redis serialization protocol  (RESP)

官網(wǎng)地址 ：

https://redis.io/docs/reference/protocol-spec/#bulk-strings  

RESP 定義了5種格式：

• 簡(jiǎn)單字符串(Simple String): 服務(wù)器用來(lái)返回簡(jiǎn)單的結(jié)果，比如"OK"。非二進(jìn)制安全，且不允許換行。

• 錯(cuò)誤信息(Error): 服務(wù)器用來(lái)返回簡(jiǎn)單的錯(cuò)誤信息，比如"ERR Invalid Synatx"。非二進(jìn)制安全，且不允許換行。

• 整數(shù)(Integer): llen、scard 等命令的返回值, 64位有符號(hào)整數(shù)

• 字符串(Bulk String): 二進(jìn)制安全字符串, 比如 get 等命令的返回值

• 數(shù)組(Array, 又稱 Multi Bulk Strings): Bulk String 數(shù)組，客戶端發(fā)送指令以及 lrange 等命令響應(yīng)的格式

5種格式通過(guò)第一個(gè)字符來(lái)區(qū)分：

• 簡(jiǎn)單字符串：以+ 開始， 如：+OK\r\n
• 錯(cuò)誤：以- 開始，如：-ERR Invalid Synatx\r\n
• 整數(shù)：以: 開始，如：:1\r\n
• 字符串：以 $ 開始，如： $3\r\nSET\r\n,3表示字符串set的字節(jié)長(zhǎng)度為3，后面跟上實(shí)際的字符串SET。 有個(gè)特例：$-1 表示 nil, 比如使用 get 命令查詢一個(gè)不存在的key時(shí)，響應(yīng)即為$-1
• 數(shù)組：以 * 開始

比如我們?cè)诿钚兄谐懙拿?code>set key value在redis-cli通過(guò)網(wǎng)絡(luò)發(fā)送給服務(wù)器端的時(shí)候，其實(shí)發(fā)送的是*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n

代碼實(shí)現(xiàn)

通過(guò)上篇文章可以，我們需要實(shí)現(xiàn)一個(gè)redisHander處理連接，本質(zhì)就是要用到本篇協(xié)議解析規(guī)則

func (t *TCPServer) handleConn(conn net.Conn) {
	// ...代碼省略...
	
	logger.Debugf("accept new conn %s", conn.RemoteAddr().String())

	// TODO :處理連接
	t.redisHander.Handle(context.Background(), conn)
}

代碼路徑：redis/handler.go

關(guān)鍵函數(shù)Handle如下：代碼思路就是啟動(dòng)一個(gè)協(xié)程parser.ParseStream(conn)負(fù)責(zé)從conn中按照\r\n為分隔符，讀取數(shù)據(jù)，并保存到chan中；然后在Handle中讀取 chan的數(shù)據(jù)，這里其實(shí)又使用到了生產(chǎn)者消費(fèi)者模型

// 該方法是不同的conn復(fù)用的方法，要做的事情就是從conn中讀取出符合RESP格式的數(shù)據(jù)；
// 然后針對(duì)消息格式，進(jìn)行不同的業(yè)務(wù)處理
func (h *RedisHandler) Handle(ctx context.Context, conn net.Conn) {
	h.activeConn.Store(conn, struct{}{})
	outChan := parser.ParseStream(conn)
	for payload := range outChan {
		if payload.Err != nil {
			// 網(wǎng)絡(luò)conn關(guān)閉
			if payload.Err == io.EOF || payload.Err == io.ErrUnexpectedEOF || strings.Contains(payload.Err.Error(), "use of closed network connection") {
				h.activeConn.Delete(conn)
				conn.Close()
				logger.Warn("client closed:" + conn.RemoteAddr().String())
				return
			}
			// 解析出錯(cuò) protocol error
			errReply := protocal.NewGenericErrReply(payload.Err.Error())
			_, err := conn.Write(errReply.ToBytes())
			if err != nil {
				h.activeConn.Delete(conn)
				conn.Close()
				logger.Warn("client closed:" + conn.RemoteAddr().String() + " err info: " + err.Error())
				return
			}
			continue
		}
		if payload.Reply == nil {
			logger.Error("empty payload")
			continue
		}
		reply, ok := payload.Reply.(*protocal.MultiBulkReply)
		if !ok {
			logger.Error("require multi bulk protocol")
			continue
		}
		logger.Debugf("%q", string(reply.ToBytes()))
		result := h.engine.Exec(conn, reply.RedisCommand)
		if result != nil {
			conn.Write(result.ToBytes())
		} else {
			conn.Write(protocal.NewUnknownErrReply().ToBytes())
		}
	}
}

查看parser.ParseStream(conn)內(nèi)部代碼，可知協(xié)議解析邏輯主要在 redis/parser.go文件中的 parse函數(shù)中，代碼注釋很清晰。

// 從r中讀取數(shù)據(jù)，將讀取的結(jié)果通過(guò) out chan 發(fā)送給外部使用（包括：正常的數(shù)據(jù)包 or 網(wǎng)絡(luò)錯(cuò)誤）
func parse(r io.Reader, out chan<- *Payload) {
	// 異?；謴?fù)，避免未知異常
	deferfunc() {
		if err := recover(); err != nil {
			logger.Error(err, string(debug.Stack()))
		}
	}()
	reader := bufio.NewReader(r)
	for {
		// 按照 \n 分隔符讀取一行數(shù)據(jù)
		line, err := reader.ReadBytes('\n')
		if err != nil { // 一般是 io.EOF錯(cuò)誤（說(shuō)明conn關(guān)閉or文件尾部）
			out <- &Payload{Err: err}
			close(out)
			return
		}
		// 讀取到的line中包括 \n 分割符
		length := len(line)
		// RESP協(xié)議是按照 \r\n 分割數(shù)據(jù)
		if length <= 2 || line[length-2] != '\r' { // 說(shuō)明是空白行，忽略
			continue
		}
		// 去掉尾部 \r\n
		line = bytes.TrimSuffix(line, []byte{'\r', '\n'})
		// 協(xié)議文檔 ：https://redis.io/docs/reference/protocol-spec/
		// The first byte in an RESP-serialized payload always identifies its type. Subsequent bytes constitute the type's contents.
		switch line[0] {
		case'*': // * 表示數(shù)組
			err := parseArrays(line, reader, out)
			if err != nil {
				out <- &Payload{Err: err}
				close(out)
				return
			}
		default:
			args := bytes.Split(line, []byte{' '})
			out <- &Payload{
				Reply: protocal.NewMultiBulkReply(args),
			}
		}
	}
}

唯一需要強(qiáng)調(diào)的一個(gè)點(diǎn)RESP協(xié)議一直強(qiáng)調(diào) 字符串(Bulk String): 二進(jìn)制安全字符串,在代碼中是如何實(shí)現(xiàn)的？？ 從conn中讀取數(shù)據(jù)，我們是按照\r\n為分隔符號(hào)獲取一串字節(jié)，那如果數(shù)據(jù)本身就帶有\r\n，那肯定就有問(wèn)題了。 例如字符串原樣輸出樣式為: $5\r\nva\r\nl\r\n $5 表示字符串長(zhǎng)度為5【va\r\nl】，所以在讀取到5的時(shí)候，我們不能繼續(xù)按照\r\n為分隔符號(hào)讀取，而是使用 io.ReadFull(reader, body)函數(shù)直接讀取5個(gè)字節(jié)

// 基于數(shù)字5 讀取 5+2 長(zhǎng)度的數(shù)據(jù)，這里的2表示\r\n
body := make([]byte, dataLen+2)
// 注意：這里直接讀取指定長(zhǎng)度的字節(jié)
_, err := io.ReadFull(reader, body)
if err != nil {
    return err
}
// 所以最終讀取到的是 hello\r\n，去掉\r\n 保存到 lines中
lines = append(lines, body[:len(body)-2])

效果展示

用官方的redis-cli 客戶端連接自己的EasyRedis服務(wù)，并發(fā)送 get easyredis 和 set easyredis 1 命令，基于 Redis序列化協(xié)議，我們可以正確的解析出命令，格式為:

*2\r\n$3\r\nget\r\n$9\r\neasyredis\r\n
*3\r\n$3\r\nset\r\n$9\r\neasyredis\r\n$1\r\n1\r\n

下篇文章就是在內(nèi)存數(shù)據(jù)庫(kù)中完成對(duì)命令的KV存儲(chǔ)（敬請(qǐng)期待）

擴(kuò)展知識(shí)

在上篇文章解析conf文件用了到了 NewScanner,本篇文章將使用NewReader從網(wǎng)絡(luò)連接中讀取數(shù)據(jù)包進(jìn)行解析

NewReader 和 NewScanner 介紹（來(lái)自ChatGPT):

在 Go 語(yǔ)言中，NewReader 和 NewScanner 分別是 bufio 包中的兩個(gè)函數(shù)，用于創(chuàng)建不同類型的讀取器。

func NewReader(rd io.Reader) *Reader

NewReader 用于創(chuàng)建一個(gè)新的 Reader 對(duì)象，該對(duì)象實(shí)現(xiàn)了 io.Reader 接口，并提供了一些額外的緩沖功能。它會(huì)使用默認(rèn)的緩沖區(qū)大?。?096 字節(jié)）。

示例：

file, err := os.Open("example.txt")
if err != nil {
    log.Fatal(err)
}
defer file.Close()
reader := bufio.NewReader(file)

這里創(chuàng)建了一個(gè)從文件中讀取的 bufio.Reader。

func NewScanner(r io.Reader) *Scanner

NewScanner 用于創(chuàng)建一個(gè)新的 Scanner 對(duì)象，該對(duì)象實(shí)現(xiàn)了 io.Scanner 接口，用于方便地從輸入源讀取數(shù)據(jù)。Scanner 對(duì)象使用默認(rèn)的 bufio.Reader 進(jìn)行緩沖。

示例：

file, err := os.Open("example.txt")
if err != nil {
    log.Fatal(err)
}
defer file.Close()

scanner := bufio.NewScanner(file)

這里創(chuàng)建了一個(gè)從文件中讀取的 bufio.Scanner。

NewReader 和 NewScanner 區(qū)別：

bufio.NewReader 返回一個(gè) bufio.Reader，該對(duì)象實(shí)現(xiàn)了 io.Reader 接口，提供了緩沖功能，適用于低層次的字節(jié)讀取。

bufio.NewScanner 返回一個(gè) bufio.Scanner，該對(duì)象實(shí)現(xiàn)了 io.Scanner 接口，提供了一些方便的方法來(lái)讀取文本數(shù)據(jù)，并且它默認(rèn)使用 bufio.Reader 進(jìn)行緩沖，適用于高層次的文本數(shù)據(jù)讀取。

選擇使用哪一個(gè)取決于你的需求。如果你需要讀取字節(jié)數(shù)據(jù)并且想要利用緩沖，可以使用 bufio.NewReader。 如果你要處理文本數(shù)據(jù)，并且想要方便地使用 Scanner 提供的方法，可以使用 bufio.NewScanner。

項(xiàng)目代碼地址: https://github.com/gofish2020/easyredis

以上就是Golang實(shí)現(xiàn)Redis網(wǎng)絡(luò)協(xié)議實(shí)例探究的詳細(xì)內(nèi)容，更多關(guān)于Golang Redis網(wǎng)絡(luò)協(xié)議的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

最新評(píng)論