GO實現Redis：GO實現記憶體資料庫（3）

来源:https://www.cnblogs.com/csgopher/archive/2023/03/24/17249335.html
-Advertisement-
Play Games

實現Redis的database層（核心層：處理命令並返回） https://github.com/csgopher/go-redis 本文涉及以下文件： dict：定義字典的一些方法 sync_dict：實現dict db：分資料庫 command：定義指令 ping，keys，string：指令 ...



  • 實現Redis的database層（核心層：處理命令並返回）
  • https://github.com/csgopher/go-redis
  • 本文涉及以下文件：
    dict：定義字典的一些方法
    sync_dict：實現dict
    db：分資料庫
    command：定義指令
    ping，keys，string：指令的具體處理邏輯
    database：單機版資料庫

datastruct/dict/dict.go

type Consumer func(key string, val interface{}) bool

type Dict interface {
   Get(key string) (val interface{}, exists bool)
   Len() int
   Put(key string, val interface{}) (result int)
   PutIfAbsent(key string, val interface{}) (result int)
   PutIfExists(key string, val interface{}) (result int)
   Remove(key string) (result int)
   ForEach(consumer Consumer)
   Keys() []string
   RandomKeys(limit int) []string
   RandomDistinctKeys(limit int) []string
   Clear()
}

Dict介面：Redis數據結構的介面。這裡我們使用sync.Map作為字典的實現，如果想用別的數據結構，換一個實現即可
Consumer：遍歷字典所有的鍵值對，返回值是布爾，true繼續遍歷，false停止遍歷

datastruct/dict/sync_dict.go

type SyncDict struct {
   m sync.Map
}

func MakeSyncDict() *SyncDict {
   return &SyncDict{}
}

func (dict *SyncDict) Get(key string) (val interface{}, exists bool) {
   val, ok := dict.m.Load(key)
   return val, ok
}

func (dict *SyncDict) Len() int {
   length := 0
   dict.m.Range(func(k, v interface{}) bool {
      length++
      return true
   })
   return length
}

func (dict *SyncDict) Put(key string, val interface{}) (result int) {
   _, existed := dict.m.Load(key)
   dict.m.Store(key, val)
   if existed {
      return 0
   }
   return 1
}

func (dict *SyncDict) PutIfAbsent(key string, val interface{}) (result int) {
   _, existed := dict.m.Load(key)
   if existed {
      return 0
   }
   dict.m.Store(key, val)
   return 1
}

func (dict *SyncDict) PutIfExists(key string, val interface{}) (result int) {
   _, existed := dict.m.Load(key)
   if existed {
      dict.m.Store(key, val)
      return 1
   }
   return 0
}

func (dict *SyncDict) Remove(key string) (result int) {
   _, existed := dict.m.Load(key)
   dict.m.Delete(key)
   if existed {
      return 1
   }
   return 0
}

func (dict *SyncDict) ForEach(consumer Consumer) {
   dict.m.Range(func(key, value interface{}) bool {
      consumer(key.(string), value)
      return true
   })
}

func (dict *SyncDict) Keys() []string {
   result := make([]string, dict.Len())
   i := 0
   dict.m.Range(func(key, value interface{}) bool {
      result[i] = key.(string)
      i++
      return true
   })
   return result
}

func (dict *SyncDict) RandomKeys(limit int) []string {
   result := make([]string, limit)
   for i := 0; i < limit; i++ {
      dict.m.Range(func(key, value interface{}) bool {
         result[i] = key.(string)
         return false
      })
   }
   return result
}

func (dict *SyncDict) RandomDistinctKeys(limit int) []string {
   result := make([]string, limit)
   i := 0
   dict.m.Range(func(key, value interface{}) bool {
      result[i] = key.(string)
      i++
      if i == limit {
         return false
      }
      return true
   })
   return result
}

func (dict *SyncDict) Clear() {
   *dict = *MakeSyncDict()
}

使用sync.Map實現Dict介面

database/db.go

type DB struct {
	index int
	data  dict.Dict
}

type ExecFunc func(db *DB, args [][]byte) resp.Reply

type CmdLine = [][]byte

func makeDB() *DB {
	db := &DB{
		data: dict.MakeSyncDict(),
	}
	return db
}

func (db *DB) Exec(c resp.Connection, cmdLine [][]byte) resp.Reply {
	cmdName := strings.ToLower(string(cmdLine[0]))
	cmd, ok := cmdTable[cmdName]
	if !ok {
		return reply.MakeErrReply("ERR unknown command '" + cmdName + "'")
	}
	if !validateArity(cmd.arity, cmdLine) {
		return reply.MakeArgNumErrReply(cmdName)
	}
	fun := cmd.executor
	return fun(db, cmdLine[1:]) // 把 set k v 中的set切掉
}

func validateArity(arity int, cmdArgs [][]byte) bool {
	argNum := len(cmdArgs)
	if arity >= 0 {
		return argNum == arity
	}
	return argNum >= -arity
}

func (db *DB) GetEntity(key string) (*database.DataEntity, bool) {
	raw, ok := db.data.Get(key)
	if !ok {
		return nil, false
	}
	entity, _ := raw.(*database.DataEntity)
	return entity, true
}

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

func (db *DB) PutIfExists(key string, entity *database.DataEntity) int {
	return db.data.PutIfExists(key, entity)
}

func (db *DB) PutIfAbsent(key string, entity *database.DataEntity) int {
	return db.data.PutIfAbsent(key, entity)
}

func (db *DB) Remove(key string) {
	db.data.Remove(key)
}

func (db *DB) Removes(keys ...string) (deleted int) {
	deleted = 0
	for _, key := range keys {
		_, exists := db.data.Get(key)
		if exists {
			db.Remove(key)
			deleted++
		}
	}
	return deleted
}

func (db *DB) Flush() {
	db.data.Clear()
}

實現Redis中的分資料庫
ExecFunc：所有Redis的指令都寫成這樣的類型
validateArity方法：

  • 定長：set k v => arity=3；
  • 變長：exists k1 k2 k3 ... => arity=-2，表示參數>=2個

database/command.go

var cmdTable = make(map[string]*command)

type command struct {
   executor ExecFunc
   arity    int 
}

func RegisterCommand(name string, executor ExecFunc, arity int) {
   name = strings.ToLower(name)
   cmdTable[name] = &command{
      executor: executor,
      arity:    arity,
   }
}

command：每一個command結構體都是一個指令，例如ping，keys等等
arity：參數數量
cmdTable：記錄所有指令和command結構體的關係
RegisterCommand：註冊指令的實現，在程式

database/ping.go

func Ping(db *DB, args [][]byte) resp.Reply {
    if len(args) == 0 {
        return &reply.PongReply{}
    } else if len(args) == 1 {
        return reply.MakeStatusReply(string(args[0]))
    } else {
        return reply.MakeErrReply("ERR wrong number of arguments for 'ping' command")
    }
}

func init() {
    RegisterCommand("ping", Ping, 1)
}

init方法：在啟動程式時就會調用這個方法，用於初始化

database/keys.go

func execDel(db *DB, args [][]byte) resp.Reply {
   keys := make([]string, len(args))
   for i, v := range args {
      keys[i] = string(v)
   }

   deleted := db.Removes(keys...)
   return reply.MakeIntReply(int64(deleted))
}

func execExists(db *DB, args [][]byte) resp.Reply {
   result := int64(0)
   for _, arg := range args {
      key := string(arg)
      _, exists := db.GetEntity(key)
      if exists {
         result++
      }
   }
   return reply.MakeIntReply(result)
}

func execFlushDB(db *DB, args [][]byte) resp.Reply {
   db.Flush()
   return &reply.OkReply{}
}

func execType(db *DB, args [][]byte) resp.Reply {
   key := string(args[0])
   entity, exists := db.GetEntity(key)
   if !exists {
      return reply.MakeStatusReply("none")
   }
   switch entity.Data.(type) {
   case []byte:
      return reply.MakeStatusReply("string")
   }
   return &reply.UnknownErrReply{}
}

func execRename(db *DB, args [][]byte) resp.Reply {
   if len(args) != 2 {
      return reply.MakeErrReply("ERR wrong number of arguments for 'rename' command")
   }
   src := string(args[0])
   dest := string(args[1])
   
   entity, ok := db.GetEntity(src)
   if !ok {
      return reply.MakeErrReply("no such key")
   }
   db.PutEntity(dest, entity)
   db.Remove(src)
   return &reply.OkReply{}
}

func execRenameNx(db *DB, args [][]byte) resp.Reply {
   src := string(args[0])
   dest := string(args[1])

   _, exist := db.GetEntity(dest)
   if exist {
      return reply.MakeIntReply(0)
   }

   entity, ok := db.GetEntity(src)
   if !ok {
      return reply.MakeErrReply("no such key")
   }
   db.Removes(src, dest)
   db.PutEntity(dest, entity)
   return reply.MakeIntReply(1)
}

func execKeys(db *DB, args [][]byte) resp.Reply {
   pattern := wildcard.CompilePattern(string(args[0]))
   result := make([][]byte, 0)
   db.data.ForEach(func(key string, val interface{}) bool {
      if pattern.IsMatch(key) {
         result = append(result, []byte(key))
      }
      return true
   })
   return reply.MakeMultiBulkReply(result)
}

func init() {
   RegisterCommand("Del", execDel, -2)
   RegisterCommand("Exists", execExists, -2)
   RegisterCommand("Keys", execKeys, 2)
   RegisterCommand("FlushDB", execFlushDB, -1)
   RegisterCommand("Type", execType, 2)
   RegisterCommand("Rename", execRename, 3)
   RegisterCommand("RenameNx", execRenameNx, 3)
}

keys.go實現以下指令：
execDel：del k1 k2 k3 ...
execExists：exist k1 k2 k3 ...
execFlushDB：flushdb
execType：type k1
execRename：rename k1 k2
execRenameNx：renamenx k1 k2
execKeys：keys（依賴lib包的工具類wildcard.go）

database/string.go

func execGet(db *DB, args [][]byte) resp.Reply {
   key := string(args[0])
   bytes, err := db.getAsString(key)
   if err != nil {
      return err
   }
   if bytes == nil {
      return &reply.NullBulkReply{}
   }
   return reply.MakeBulkReply(bytes)
}

func (db *DB) getAsString(key string) ([]byte, reply.ErrorReply) {
   entity, ok := db.GetEntity(key)
   if !ok {
      return nil, nil
   }
   bytes, ok := entity.Data.([]byte)
   if !ok {
      return nil, &reply.WrongTypeErrReply{}
   }
   return bytes, nil
}

func execSet(db *DB, args [][]byte) resp.Reply {
   key := string(args[0])
   value := args[1]
   entity := &database.DataEntity{
      Data: value,
   }
   db.PutEntity(key, entity)
   return &reply.OkReply{}
}

func execSetNX(db *DB, args [][]byte) resp.Reply {
   key := string(args[0])
   value := args[1]
   entity := &database.DataEntity{
      Data: value,
   }
   result := db.PutIfAbsent(key, entity)
   return reply.MakeIntReply(int64(result))
}

func execGetSet(db *DB, args [][]byte) resp.Reply {
   key := string(args[0])
   value := args[1]

   entity, exists := db.GetEntity(key)
   db.PutEntity(key, &database.DataEntity{Data: value})
   if !exists {
      return reply.MakeNullBulkReply()
   }
   old := entity.Data.([]byte)
   return reply.MakeBulkReply(old)
}

func execStrLen(db *DB, args [][]byte) resp.Reply {
   key := string(args[0])
   entity, exists := db.GetEntity(key)
   if !exists {
      return reply.MakeNullBulkReply()
   }
   old := entity.Data.([]byte)
   return reply.MakeIntReply(int64(len(old)))
}

func init() {
   RegisterCommand("Get", execGet, 2)
   RegisterCommand("Set", execSet, -3)
   RegisterCommand("SetNx", execSetNX, 3)
   RegisterCommand("GetSet", execGetSet, 3)
   RegisterCommand("StrLen", execStrLen, 2)
}

string.go實現以下指令：
execGet：get k1
execSet：set k v
execSetNX：setnex k v
execGetSet：getset k v 返回舊值
execStrLen：strlen k

database/database.go

type Database struct {
   dbSet []*DB
}

func NewDatabase() *Database {
   mdb := &Database{}
   if config.Properties.Databases == 0 {
      config.Properties.Databases = 16
   }
   mdb.dbSet = make([]*DB, config.Properties.Databases)
   for i := range mdb.dbSet {
      singleDB := makeDB()
      singleDB.index = i
      mdb.dbSet[i] = singleDB
   }
   return mdb
}

func (mdb *Database) Exec(c resp.Connection, cmdLine [][]byte) (result resp.Reply) {
   defer func() {
      if err := recover(); err != nil {
         logger.Warn(fmt.Sprintf("error occurs: %v\n%s", err, string(debug.Stack())))
      }
   }()

   cmdName := strings.ToLower(string(cmdLine[0]))
   if cmdName == "select" {
      if len(cmdLine) != 2 {
         return reply.MakeArgNumErrReply("select")
      }
      return execSelect(c, mdb, cmdLine[1:])
   }
   dbIndex := c.GetDBIndex()
   selectedDB := mdb.dbSet[dbIndex]
   return selectedDB.Exec(c, cmdLine)
}

func execSelect(c resp.Connection, mdb *Database, args [][]byte) resp.Reply {
   dbIndex, err := strconv.Atoi(string(args[0]))
   if err != nil {
      return reply.MakeErrReply("ERR invalid DB index")
   }
   if dbIndex >= len(mdb.dbSet) {
      return reply.MakeErrReply("ERR DB index is out of range")
   }
   c.SelectDB(dbIndex)
   return reply.MakeOkReply()
}

func (mdb *Database) Close() {
}

func (mdb *Database) AfterClientClose(c resp.Connection) {
}

Database：一組db的集合
Exec：執行切換db指令或者其他指令
execSelect方法：選擇db（指令：select 2）

resp/handler/handler.go

import (
	database2 "go-redis/database"
)

func MakeHandler() *RespHandler {
   var db database.Database
   db = database2.NewDatabase()
   return &RespHandler{
      db: db,
   }
}

修改實現協議層handler的database實現

架構小結

TCP層服務TCP的連接，然後將連接交給RESP協議層的handler，handler監聽客戶端的連接，將指令解析後發給管道，管道轉給database層（database/database.go），核心層根據命令類型執行不同的方法，然後返回。


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • Django筆記七之ManyToMany和OneToOne介紹
    該工程為在保存時執行開發的功能，函數入口點ufput。其他還有新建、打開、另存等都可以加入開發的操作，具體看UF_EXIT下的介紹。 用戶出口是一個可選特性，允許你在NX中某些預定義的位置(或出口)自動運行Open C API程式。如果你進入其中一個出口，NX會檢查你是否定義了指向Open C AP ...
  • 商城網站畢業設計（ Python +Vue）
    網站介紹 基於 python 開發的電子商城網站，平臺採用 B/S 結構，後端採用主流的 Python 語言進行開發，前端採用主流的 Vue.js 進行開發。這是給師弟開發的畢業設計。 整個平臺包括前臺和後臺兩個部分。 前臺功能包括：首頁、商品詳情頁、用戶中心模塊。 後臺功能包括：總覽、訂單管理、商 ...
  • day12-SpringBoot資料庫操作
    SpringBoot資料庫操作 1.JDBC+HikariDataSource 在SpringBoot 2.x項目中，預設使用Hikari連接池管理數據源。相比於傳統的 C3P0 、DBCP、Tomcat jdbc 等連接池更加優秀。 當項目pom.xml引入spring-boot-starter- ...
  • odoo 開發入門教程系列-一個新應用
    一個新應用 房地產廣告模塊 假設需要開發一個房地產模塊，該模塊覆蓋未包含在標準模塊集中特定業務領域。 以下為包含一些廣告的主列表視圖 form視圖頂層區域概括了房產的重要信息，比如name，Property Type, Postcode等等。 列表記錄詳情頁中，第一個tab包含了房產的描述信息，比如 ...
  • Java生產者消費者
    生產者消費者問題 簡介 生產者消費者模式並不是GOF提出的23種設計模式之一，23種設計模式都是建立在面向對象的基礎之上的，但其實面向過程的編程中也有很多高效的編程模式，生產者消費者模式便是其中之一，它是我們編程過程中最常用的一種設計模式。 在實際的軟體開發過程中，經常會碰到如下場景：某個模塊負責產 ...
  • Teamcenter_NX集成開發：通過NXOpen查詢零組件是否存在
    之前用過NXOpen PDM的命名空間下的類，現在記錄一下通過PDM命名空間下的類查詢Teamcenter零組件的信息，也可以用來判斷該零組件是否存在。 1-該工程為DLL工程，直接在NX界面調用，所以直接獲取NXSession。 2-查詢函數advanced用到的查詢為：__NX_STD_ANY_ ...
  • 9個都要瞭解的單行Python代碼
    本文已收錄至Github，推薦閱讀 👉 Java隨想錄 微信公眾號：Java隨想錄 摘要 Redis是一款性能強勁的記憶體資料庫，但是在使用過程中，我們可能會遇到Big Key問題，這個問題就是Redis中某個key的value過大，所以Big Key問題本質是Big Value問題，導致Redis ...
  • PLG SaaS 產品 Figma 商業模式拆解
    9 月 15 日，Figma 的 CEO Dylan Field 發佈消息：今天，Figma 宣佈接受 Adobe 的收購... Adobe 以約 200 億美元收購 Figma，這也是 Adobe 該公司在其歷史上的最大一筆收購。那是什麼樣的魔力，讓 Figma 被 Adobe 收購呢？下麵以定位 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類