GO实现Redis：GO实现内存数据库（3）

滴水恩情  金牌会员 | 2023-4-4 14:12:30 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 666|帖子 666|积分 1998


datastruct/dict/dict.go
  1. type Consumer func(key string, val interface{}) bool
  3. type Dict interface {
  4.    Get(key string) (val interface{}, exists bool)
  5.    Len() int
  6.    Put(key string, val interface{}) (result int)
  7.    PutIfAbsent(key string, val interface{}) (result int)
  8.    PutIfExists(key string, val interface{}) (result int)
  9.    Remove(key string) (result int)
  10.    ForEach(consumer Consumer)
  11.    Keys() []string
  12.    RandomKeys(limit int) []string
  13.    RandomDistinctKeys(limit int) []string
  14.    Clear()
  15. }
复制代码
Dict接口：Redis数据结构的接口。这里我们使用sync.Map作为字典的实现，如果想用别的数据结构，换一个实现即可
Consumer：遍历字典所有的键值对，返回值是布尔，true继续遍历，false停止遍历


datastruct/dict/sync_dict.go
  1. type SyncDict struct {
  2.    m sync.Map
  3. }
  5. func MakeSyncDict() *SyncDict {
  6.    return &SyncDict{}
  7. }
  9. func (dict *SyncDict) Get(key string) (val interface{}, exists bool) {
  10.    val, ok := dict.m.Load(key)
  11.    return val, ok
  12. }
  14. func (dict *SyncDict) Len() int {
  15.    length := 0
  16.    dict.m.Range(func(k, v interface{}) bool {
  17.       length++
  18.       return true
  19.    })
  20.    return length
  21. }
  23. func (dict *SyncDict) Put(key string, val interface{}) (result int) {
  24.    _, existed := dict.m.Load(key)
  25.    dict.m.Store(key, val)
  26.    if existed {
  27.       return 0
  28.    }
  29.    return 1
  30. }
  32. func (dict *SyncDict) PutIfAbsent(key string, val interface{}) (result int) {
  33.    _, existed := dict.m.Load(key)
  34.    if existed {
  35.       return 0
  36.    }
  37.    dict.m.Store(key, val)
  38.    return 1
  39. }
  41. func (dict *SyncDict) PutIfExists(key string, val interface{}) (result int) {
  42.    _, existed := dict.m.Load(key)
  43.    if existed {
  44.       dict.m.Store(key, val)
  45.       return 1
  46.    }
  47.    return 0
  48. }
  50. func (dict *SyncDict) Remove(key string) (result int) {
  51.    _, existed := dict.m.Load(key)
  52.    dict.m.Delete(key)
  53.    if existed {
  54.       return 1
  55.    }
  56.    return 0
  57. }
  59. func (dict *SyncDict) ForEach(consumer Consumer) {
  60.    dict.m.Range(func(key, value interface{}) bool {
  61.       consumer(key.(string), value)
  62.       return true
  63.    })
  64. }
  66. func (dict *SyncDict) Keys() []string {
  67.    result := make([]string, dict.Len())
  68.    i := 0
  69.    dict.m.Range(func(key, value interface{}) bool {
  70.       result[i] = key.(string)
  71.       i++
  72.       return true
  73.    })
  74.    return result
  75. }
  77. func (dict *SyncDict) RandomKeys(limit int) []string {
  78.    result := make([]string, limit)
  79.    for i := 0; i < limit; i++ {
  80.       dict.m.Range(func(key, value interface{}) bool {
  81.          result[i] = key.(string)
  82.          return false
  83.       })
  84.    }
  85.    return result
  86. }
  88. func (dict *SyncDict) RandomDistinctKeys(limit int) []string {
  89.    result := make([]string, limit)
  90.    i := 0
  91.    dict.m.Range(func(key, value interface{}) bool {
  92.       result[i] = key.(string)
  93.       i++
  94.       if i == limit {
  95.          return false
  96.       }
  97.       return true
  98.    })
  99.    return result
  100. }
  102. func (dict *SyncDict) Clear() {
  103.    *dict = *MakeSyncDict()
  104. }
复制代码
使用sync.Map实现Dict接口


database/db.go
  1. type DB struct {
  2.         index int
  3.         data  dict.Dict
  4. }
  6. type ExecFunc func(db *DB, args [][]byte) resp.Reply
  8. type CmdLine = [][]byte
  10. func makeDB() *DB {
  11.         db := &DB{
  12.                 data: dict.MakeSyncDict(),
  13.         }
  14.         return db
  15. }
  17. func (db *DB) Exec(c resp.Connection, cmdLine [][]byte) resp.Reply {
  18.         cmdName := strings.ToLower(string(cmdLine[0]))
  19.         cmd, ok := cmdTable[cmdName]
  20.         if !ok {
  21.                 return reply.MakeErrReply("ERR unknown command '" + cmdName + "'")
  22.         }
  23.         if !validateArity(cmd.arity, cmdLine) {
  24.                 return reply.MakeArgNumErrReply(cmdName)
  25.         }
  26.         fun := cmd.executor
  27.         return fun(db, cmdLine[1:]) // 把 set k v 中的set切掉
  28. }
  30. func validateArity(arity int, cmdArgs [][]byte) bool {
  31.         argNum := len(cmdArgs)
  32.         if arity >= 0 {
  33.                 return argNum == arity
  34.         }
  35.         return argNum >= -arity
  36. }
  38. func (db *DB) GetEntity(key string) (*database.DataEntity, bool) {
  39.         raw, ok := db.data.Get(key)
  40.         if !ok {
  41.                 return nil, false
  42.         }
  43.         entity, _ := raw.(*database.DataEntity)
  44.         return entity, true
  45. }
  47. func (db *DB) PutEntity(key string, entity *database.DataEntity) int {
  48.         return db.data.Put(key, entity)
  49. }
  51. func (db *DB) PutIfExists(key string, entity *database.DataEntity) int {
  52.         return db.data.PutIfExists(key, entity)
  53. }
  55. func (db *DB) PutIfAbsent(key string, entity *database.DataEntity) int {
  56.         return db.data.PutIfAbsent(key, entity)
  57. }
  59. func (db *DB) Remove(key string) {
  60.         db.data.Remove(key)
  61. }
  63. func (db *DB) Removes(keys ...string) (deleted int) {
  64.         deleted = 0
  65.         for _, key := range keys {
  66.                 _, exists := db.data.Get(key)
  67.                 if exists {
  68.                         db.Remove(key)
  69.                         deleted++
  70.                 }
  71.         }
  72.         return deleted
  73. }
  75. func (db *DB) Flush() {
  76.         db.data.Clear()
  77. }
复制代码
实现Redis中的分数据库
ExecFunc：所有Redis的指令都写成这样的类型
validateArity方法：

  • 定长：set k v => arity=3；
  • 变长：exists k1 k2 k3 ... => arity=-2，表示参数>=2个
database/command.go
  1. var cmdTable = make(map[string]*command)
  3. type command struct {
  4.    executor ExecFunc
  5.    arity    int
  6. }
  8. func RegisterCommand(name string, executor ExecFunc, arity int) {
  9.    name = strings.ToLower(name)
  10.    cmdTable[name] = &command{
  11.       executor: executor,
  12.       arity:    arity,
  13.    }
  14. }
复制代码
command：每一个command结构体都是一个指令，例如ping，keys等等
arity：参数数量
cmdTable：记录所有指令和command结构体的关系
RegisterCommand：注册指令的实现，在程序


database/ping.go
  1. func Ping(db *DB, args [][]byte) resp.Reply {
  2.     if len(args) == 0 {
  3.         return &reply.PongReply{}
  4.     } else if len(args) == 1 {
  5.         return reply.MakeStatusReply(string(args[0]))
  6.     } else {
  7.         return reply.MakeErrReply("ERR wrong number of arguments for 'ping' command")
  8.     }
  9. }
  11. func init() {
  12.     RegisterCommand("ping", Ping, 1)
  13. }
复制代码
init方法：在启动程序时就会调用这个方法，用于初始化


database/keys.go
  1. func execDel(db *DB, args [][]byte) resp.Reply {
  2.    keys := make([]string, len(args))
  3.    for i, v := range args {
  4.       keys[i] = string(v)
  5.    }
  7.    deleted := db.Removes(keys...)
  8.    return reply.MakeIntReply(int64(deleted))
  9. }
  11. func execExists(db *DB, args [][]byte) resp.Reply {
  12.    result := int64(0)
  13.    for _, arg := range args {
  14.       key := string(arg)
  15.       _, exists := db.GetEntity(key)
  16.       if exists {
  17.          result++
  18.       }
  19.    }
  20.    return reply.MakeIntReply(result)
  21. }
  23. func execFlushDB(db *DB, args [][]byte) resp.Reply {
  24.    db.Flush()
  25.    return &reply.OkReply{}
  26. }
  28. func execType(db *DB, args [][]byte) resp.Reply {
  29.    key := string(args[0])
  30.    entity, exists := db.GetEntity(key)
  31.    if !exists {
  32.       return reply.MakeStatusReply("none")
  33.    }
  34.    switch entity.Data.(type) {
  35.    case []byte:
  36.       return reply.MakeStatusReply("string")
  37.    }
  38.    return &reply.UnknownErrReply{}
  39. }
  41. func execRename(db *DB, args [][]byte) resp.Reply {
  42.    if len(args) != 2 {
  43.       return reply.MakeErrReply("ERR wrong number of arguments for 'rename' command")
  44.    }
  45.    src := string(args[0])
  46.    dest := string(args[1])
  47.    
  48.    entity, ok := db.GetEntity(src)
  49.    if !ok {
  50.       return reply.MakeErrReply("no such key")
  51.    }
  52.    db.PutEntity(dest, entity)
  53.    db.Remove(src)
  54.    return &reply.OkReply{}
  55. }
  57. func execRenameNx(db *DB, args [][]byte) resp.Reply {
  58.    src := string(args[0])
  59.    dest := string(args[1])
  61.    _, exist := db.GetEntity(dest)
  62.    if exist {
  63.       return reply.MakeIntReply(0)
  64.    }
  66.    entity, ok := db.GetEntity(src)
  67.    if !ok {
  68.       return reply.MakeErrReply("no such key")
  69.    }
  70.    db.Removes(src, dest)
  71.    db.PutEntity(dest, entity)
  72.    return reply.MakeIntReply(1)
  73. }
  75. func execKeys(db *DB, args [][]byte) resp.Reply {
  76.    pattern := wildcard.CompilePattern(string(args[0]))
  77.    result := make([][]byte, 0)
  78.    db.data.ForEach(func(key string, val interface{}) bool {
  79.       if pattern.IsMatch(key) {
  80.          result = append(result, []byte(key))
  81.       }
  82.       return true
  83.    })
  84.    return reply.MakeMultiBulkReply(result)
  85. }
  87. func init() {
  88.    RegisterCommand("Del", execDel, -2)
  89.    RegisterCommand("Exists", execExists, -2)
  90.    RegisterCommand("Keys", execKeys, 2)
  91.    RegisterCommand("FlushDB", execFlushDB, -1)
  92.    RegisterCommand("Type", execType, 2)
  93.    RegisterCommand("Rename", execRename, 3)
  94.    RegisterCommand("RenameNx", execRenameNx, 3)
  95. }
复制代码
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
  1. func execGet(db *DB, args [][]byte) resp.Reply {
  2.    key := string(args[0])
  3.    bytes, err := db.getAsString(key)
  4.    if err != nil {
  5.       return err
  6.    }
  7.    if bytes == nil {
  8.       return &reply.NullBulkReply{}
  9.    }
  10.    return reply.MakeBulkReply(bytes)
  11. }
  13. func (db *DB) getAsString(key string) ([]byte, reply.ErrorReply) {
  14.    entity, ok := db.GetEntity(key)
  15.    if !ok {
  16.       return nil, nil
  17.    }
  18.    bytes, ok := entity.Data.([]byte)
  19.    if !ok {
  20.       return nil, &reply.WrongTypeErrReply{}
  21.    }
  22.    return bytes, nil
  23. }
  25. func execSet(db *DB, args [][]byte) resp.Reply {
  26.    key := string(args[0])
  27.    value := args[1]
  28.    entity := &database.DataEntity{
  29.       Data: value,
  30.    }
  31.    db.PutEntity(key, entity)
  32.    return &reply.OkReply{}
  33. }
  35. func execSetNX(db *DB, args [][]byte) resp.Reply {
  36.    key := string(args[0])
  37.    value := args[1]
  38.    entity := &database.DataEntity{
  39.       Data: value,
  40.    }
  41.    result := db.PutIfAbsent(key, entity)
  42.    return reply.MakeIntReply(int64(result))
  43. }
  45. func execGetSet(db *DB, args [][]byte) resp.Reply {
  46.    key := string(args[0])
  47.    value := args[1]
  49.    entity, exists := db.GetEntity(key)
  50.    db.PutEntity(key, &database.DataEntity{Data: value})
  51.    if !exists {
  52.       return reply.MakeNullBulkReply()
  53.    }
  54.    old := entity.Data.([]byte)
  55.    return reply.MakeBulkReply(old)
  56. }
  58. func execStrLen(db *DB, args [][]byte) resp.Reply {
  59.    key := string(args[0])
  60.    entity, exists := db.GetEntity(key)
  61.    if !exists {
  62.       return reply.MakeNullBulkReply()
  63.    }
  64.    old := entity.Data.([]byte)
  65.    return reply.MakeIntReply(int64(len(old)))
  66. }
  68. func init() {
  69.    RegisterCommand("Get", execGet, 2)
  70.    RegisterCommand("Set", execSet, -3)
  71.    RegisterCommand("SetNx", execSetNX, 3)
  72.    RegisterCommand("GetSet", execGetSet, 3)
  73.    RegisterCommand("StrLen", execStrLen, 2)
  74. }
复制代码
string.go实现以下指令：
execGet：get k1
execSet：set k v
execSetNX：setnex k v
execGetSet：getset k v 返回旧值
execStrLen：strlen k


database/database.go
  1. type Database struct {
  2.    dbSet []*DB
  3. }
  5. func NewDatabase() *Database {
  6.    mdb := &Database{}
  7.    if config.Properties.Databases == 0 {
  8.       config.Properties.Databases = 16
  9.    }
  10.    mdb.dbSet = make([]*DB, config.Properties.Databases)
  11.    for i := range mdb.dbSet {
  12.       singleDB := makeDB()
  13.       singleDB.index = i
  14.       mdb.dbSet[i] = singleDB
  15.    }
  16.    return mdb
  17. }
  19. func (mdb *Database) Exec(c resp.Connection, cmdLine [][]byte) (result resp.Reply) {
  20.    defer func() {
  21.       if err := recover(); err != nil {
  22.          logger.Warn(fmt.Sprintf("error occurs: %v\n%s", err, string(debug.Stack())))
  23.       }
  24.    }()
  26.    cmdName := strings.ToLower(string(cmdLine[0]))
  27.    if cmdName == "select" {
  28.       if len(cmdLine) != 2 {
  29.          return reply.MakeArgNumErrReply("select")
  30.       }
  31.       return execSelect(c, mdb, cmdLine[1:])
  32.    }
  33.    dbIndex := c.GetDBIndex()
  34.    selectedDB := mdb.dbSet[dbIndex]
  35.    return selectedDB.Exec(c, cmdLine)
  36. }
  38. func execSelect(c resp.Connection, mdb *Database, args [][]byte) resp.Reply {
  39.    dbIndex, err := strconv.Atoi(string(args[0]))
  40.    if err != nil {
  41.       return reply.MakeErrReply("ERR invalid DB index")
  42.    }
  43.    if dbIndex >= len(mdb.dbSet) {
  44.       return reply.MakeErrReply("ERR DB index is out of range")
  45.    }
  46.    c.SelectDB(dbIndex)
  47.    return reply.MakeOkReply()
  48. }
  50. func (mdb *Database) Close() {
  51. }
  53. func (mdb *Database) AfterClientClose(c resp.Connection) {
  54. }
复制代码
Database：一组db的集合
Exec：执行切换db指令或者其他指令
execSelect方法：选择db（指令：select 2）


resp/handler/handler.go
  1. import (
  2.         database2 "go-redis/database"
  3. )
  5. func MakeHandler() *RespHandler {
  6.    var db database.Database
  7.    db = database2.NewDatabase()
  8.    return &RespHandler{
  9.       db: db,
  10.    }
  11. }
复制代码
修改实现协议层handler的database实现
架构小结

TCP层服务TCP的连接，然后将连接交给RESP协议层的handler，handler监听客户端的连接，将指令解析后发给管道，管道转给database层（database/database.go），核心层根据命令类型执行不同的方法，然后返回。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

滴水恩情

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表