Golang實現(xiàn)自己的Redis數(shù)據(jù)庫內(nèi)存實例探究

更新時間：2024年01月24日 10:16:58 作者：紹納 nullbody筆記

這篇文章主要為大家介紹了Golang實現(xiàn)自己的Redis數(shù)據(jù)庫內(nèi)存實例探究,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪

引言

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

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

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

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

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

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

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

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

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

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

[ ] easyredis之連接池

[ ] easyredis之分布式集群存儲

EasyRedis之內(nèi)存數(shù)據(jù)庫篇

上篇文章已經(jīng)可以解析出Redis serialization protocol,本篇基于解析出來的命令，進(jìn)行代碼處理過程：這里以5個常用命令作為本篇文章的切入口：命令官方文檔 https://redis.io/commands

# ping服務(wù)器
PING [message]
# 授權(quán)密碼設(shè)置
AUTH <password>
# 選擇數(shù)據(jù)庫
SELECT index
# 設(shè)置key
SET key value [NX | XX]  [EX seconds | PX milliseconds]
# 獲取key
GET key

ping服務(wù)器

代碼路徑: engine/engine.go這個功能算是小試牛刀的小功能，讓大家對基本的套路有個簡單的認(rèn)識

// redisCommand 待執(zhí)行的命令  protocal.Reply 執(zhí)行結(jié)果
func (e *Engine) Exec(c *connection.KeepConnection, redisCommand [][]byte) (result protocal.Reply) {
	 //... 省略...
	commandName := strings.ToLower(string(redisCommand[0]))
	if commandName == "ping" { // https://redis.io/commands/ping/
		return Ping(redisCommand[1:])
	}
	 //... 省略...
}

Exec函數(shù)就是進(jìn)行命令處理的總?cè)肟诤瘮?shù)，通過從協(xié)議中解析出來的redisCommand，我們可以提取出命令名commandName變量，然后在Ping(redisCommand[1:])函數(shù)中進(jìn)行邏輯處理。

func Ping(redisArgs [][]byte) protocal.Reply {

	iflen(redisArgs) == 0 { // 不帶參數(shù)
		return protocal.NewPONGReply()
	} elseiflen(redisArgs) == 1 { // 帶參數(shù)1個
		return protocal.NewBulkReply(redisArgs[0])
	}
    // 否則，回復(fù)命令格式錯誤
	return protocal.NewArgNumErrReply("ping")
}

Ping函數(shù)的本質(zhì)就是基于PING [message]這個redis命令的基本格式，進(jìn)行不同的數(shù)據(jù)響應(yīng)。

這里建議大家看下Ping命令的文檔 https://redis.io/commands/ping/

以上圖為例

如果是直接的PING命令，后面不帶參數(shù)，我們要回復(fù)PONG
如果帶參"Hello world"，我們原樣回復(fù)Hello world
如果帶了兩個參數(shù)hello和world,直接回復(fù)錯誤。

有了這個處理套路，那么其他的命令也可依葫蘆畫瓢了。

授權(quán)密碼設(shè)置

啟動redis服務(wù)的時候，如果有設(shè)定需要密碼，那么客戶端連接上來以后，需要先執(zhí)行一次 Auth password的授權(quán)命令

// redisCommand 待執(zhí)行的命令  protocal.Reply 執(zhí)行結(jié)果
func (e *Engine) Exec(c *connection.KeepConnection, redisCommand [][]byte) (result protocal.Reply) {

	//... 省略...
	commandName := strings.ToLower(string(redisCommand[0]))
	if commandName == "auth" {
		return Auth(c, redisCommand[1:])
	}
	// 校驗密碼
	if !checkPasswd(c) {
		return protocal.NewGenericErrReply("Authentication required")
	}

	//... 省略...
}

服務(wù)器接收到命令后，依據(jù)commandName的變量值為auth，則執(zhí)行 Auth(c, redisCommand[1:])函數(shù)

func Auth(c *connection.KeepConnection, redisArgs [][]byte) protocal.Reply {
	iflen(redisArgs) != 1 {
		return protocal.NewArgNumErrReply("auth")
	}

	if conf.GlobalConfig.RequirePass == "" {
		return protocal.NewGenericErrReply("No authorization is required")
	}

	password := string(redisArgs[0])
	if conf.GlobalConfig.RequirePass != password {
		return protocal.NewGenericErrReply("Auth failed, password is wrong")
	}

	c.SetPassword(password)
	return protocal.NewOkReply()
}

這里的解析過程我們是按照 AUTH <password>這個命令格式進(jìn)行解析，解析出來密碼以后，我們需要將密碼保存在c *connection.KeepConnection對象的成員變量中。這里就類似session的原理，存儲以后，當(dāng)前連接接下來的命令就不需要繼續(xù)帶上密碼了。在每次處理其他命令之前，校驗下當(dāng)前連接的密碼是否有效：

func checkPasswd(c *connection.KeepConnection) bool {
	// 如果沒有配置密碼
	if conf.GlobalConfig.RequirePass == "" {
		returntrue
	}
	// 密碼是否一致
	return c.GetPassword() == conf.GlobalConfig.RequirePass
}

選擇數(shù)據(jù)庫

這個命令雖然用的比較少，但是這個涉及到服務(wù)端結(jié)構(gòu)的設(shè)計。redis的服務(wù)端是支持多個數(shù)據(jù)庫，每個數(shù)據(jù)庫就是一個CRUD的基本存儲單元，不同的數(shù)據(jù)庫(存儲單元)之間的數(shù)據(jù)是不共享的。默認(rèn)情況下，我們使用的都是select 0數(shù)據(jù)庫。

代碼結(jié)構(gòu)如下圖：（這個圖很重要，配合代碼好好理解下）

我們需要在Engine結(jié)構(gòu)體中創(chuàng)建多個 *DB對象

func NewEngine() *Engine {
	engine := &Engine{}
	// 多個dbSet
	engine.dbSet = make([]*atomic.Value, conf.GlobalConfig.Databases)
	for i := 0; i < conf.GlobalConfig.Databases; i++ {
		// 創(chuàng)建 *db
		db := newDB()
		db.SetIndex(i)
		// 保存到 atomic.Value中
		dbset := &atomic.Value{}
		dbset.Store(db)
		// 賦值到 dbSet中
		engine.dbSet[i] = dbset
	}
	return engine
}

在用戶端發(fā)送來 select index 命令，服務(wù)端需要記錄下來當(dāng)前連接選中的數(shù)據(jù)庫索引

// redisCommand 待執(zhí)行的命令  protocal.Reply 執(zhí)行結(jié)果
func (e *Engine) Exec(c *connection.KeepConnection, redisCommand [][]byte) (result protocal.Reply) {
	//....忽略....
	// 基礎(chǔ)命令
	switch commandName {
	case"select": // 表示當(dāng)前連接，要選中哪個db https://redis.io/commands/select/
		return execSelect(c, redisCommand[1:])
	}
	//....忽略....
}
// 這里會對 選中的索引 越界的判斷，如果一切都正常，就保存到 c *connection.KeepConnection 連接的成員變量 index
func execSelect(c *connection.KeepConnection, redisArgs [][]byte) protocal.Reply {
	iflen(redisArgs) != 1 {
		return protocal.NewArgNumErrReply("select")
	}
	dbIndex, err := strconv.ParseInt(string(redisArgs[0]), 10, 64)
	if err != nil {
		return protocal.NewGenericErrReply("invaild db index")
	}
	if dbIndex < 0 || dbIndex >= int64(conf.GlobalConfig.Databases) {
		return protocal.NewGenericErrReply("db index out of range")
	}
	c.SetDBIndex(int(dbIndex))
	return protocal.NewOkReply()
}

設(shè)置key

在用戶要求執(zhí)行 set key value命令的時候，我們需要先選中執(zhí)行功能*DB對象，就是上面的select index命令要求選中的對象，默認(rèn)是0號

// redisCommand 待執(zhí)行的命令  protocal.Reply 執(zhí)行結(jié)果
func (e *Engine) Exec(c *connection.KeepConnection, redisCommand [][]byte) (result protocal.Reply) {

	//....忽略....

	// redis 命令處理
	dbIndex := c.GetDBIndex()
	logger.Debugf("db index:%d", dbIndex)
	db, errReply := e.selectDB(dbIndex)
	if errReply != nil {
		return errReply
	}
	return db.Exec(c, redisCommand)
}

可以看到，最終代碼執(zhí)行execNormalCommand函數(shù)，該函數(shù)會從命令注冊中心獲取命令的執(zhí)行函數(shù)

func (db *DB) Exec(c *connection.KeepConnection, redisCommand [][]byte) protocal.Reply {
	return db.execNormalCommand(c, redisCommand)
}
func (db *DB) execNormalCommand(c *connection.KeepConnection, redisCommand [][]byte) protocal.Reply {
	cmdName := strings.ToLower(string(redisCommand[0]))
	// 從命令注冊中心，獲取命令的執(zhí)行函數(shù)
	command, ok := commandCenter[cmdName]
	if !ok {
		return protocal.NewGenericErrReply("unknown command '" + cmdName + "'")
	}
	fun := command.execFunc
	return fun(db, redisCommand[1:])
}

最終 set命令的實際執(zhí)行函數(shù)代碼路徑為engine/string.go中的func cmdSet(db *DB, args [][]byte) protocal.Reply函數(shù)。代碼的的本質(zhì)其實還是解析字符串，按照官方文檔https://redis.io/commands/set/ 要求的格式獲取對應(yīng)的參數(shù)，執(zhí)行數(shù)據(jù)的存儲db.PutEntity(key, &entity)。

func (db *DB) PutEntity(key string, entity *payload.DataEntity) int {
	return db.dataDict.Put(key, entity)
}

dataDict是*ConcurrentDict類型的并發(fā)安全的字典

// 并發(fā)安全的字典
type ConcurrentDict struct {
	shds  []*shard      // 底層shard切片
	mask  uint32// 掩碼
	count *atomic.Int32 // 元素個數(shù)
}

ConcurrentDict通過分片的模式，將數(shù)據(jù)分散在不同的*shard對象中,shard的本質(zhì)就是map+讀寫鎖mu

type shard struct {
	m  map[string]interface{}
	mu sync.RWMutex
}

所以內(nèi)存數(shù)據(jù)庫的本質(zhì)就是操作map

key就是set命令的key
value我們額外包裝了一個DataEntity對象，將實際的值保存在 RedisObject

type DataEntity struct {
	RedisObject interface{} // 字符串 跳表 鏈表 quicklist 集合 etc...
}

代碼中已經(jīng)注釋的很清晰，建議直接看代碼

獲取key

代碼只是額外多調(diào)用了幾層函數(shù)，本質(zhì)就是調(diào)用db.dataDict.Get(key) 函數(shù)，其實又是*ConcurrentDict,代碼可能感覺有點(diǎn)繞，把上面的代碼結(jié)構(gòu)圖好好理解一下。

func cmdGet(db *DB, args [][]byte) protocal.Reply {
	iflen(args) != 1 {
		return protocal.NewSyntaxErrReply()
	}
	key := string(args[0])
	bytes, reply := db.getStringObject(key)
	if reply != nil {
		return reply
	}
	return protocal.NewBulkReply(bytes)
}
// 獲取底層存儲對象【字節(jié)流】
func (db *DB) getStringObject(key string) ([]byte, protocal.Reply) {
	payload, exist := db.GetEntity(key)
	if !exist {
		returnnil, protocal.NewNullBulkReply()
	}
	// 判斷底層對象是否為【字節(jié)流】
	bytes, ok := payload.RedisObject.([]byte)
	if !ok {
		returnnil, protocal.NewWrongTypeErrReply()
	}
	return bytes, nil
}
// 獲取內(nèi)存中的數(shù)據(jù)
func (db *DB) GetEntity(key string) (*payload.DataEntity, bool) {
	// key 不存在
	val, exist := db.dataDict.Get(key)
	if !exist {
		returnnil, false
	}
	dataEntity, ok := val.(*payload.DataEntity)
	if !ok {
		returnnil, false
	}
	return dataEntity, true
}