second commit 实现数据库内部数据结构以及若干相应redis命令

Author: Reditiny

database/command.go

@@ -0,0 +1,14 @@
+package database
+
+import "redis-based-on-go/redis/protocol"
+
+// commandTable 记录所有客户端命令
+var commandTable = make(map[string]*command)
+
+// ExecFunc 命令执行函数
+type ExecFunc func(db *DB, args [][]byte) protocol.Reply
+
+type command struct {
+	name     string
+	executor ExecFunc
+}

database/concurrent_map.go

@@ -0,0 +1,104 @@
+package database
+
+import (
+	"sync"
+)
+
+const (
+	TABLE_COUNT = 16
+	PRIME_32    = uint32(16777619)
+)
+
+// DB 中 key-value 的并发安全结构
+type concurrentMap struct {
+	tables      []*table
+	memberCount int
+	tableCount  int
+}
+
+type table struct {
+	m     map[string]interface{}
+	mutex sync.RWMutex
+}
+
+// makeConcurrentMap makes a new concurrent map
+func makeConcurrentMap() *concurrentMap {
+	cm := &concurrentMap{
+		tables:     make([]*table, TABLE_COUNT),
+		tableCount: TABLE_COUNT,
+	}
+	for i := 0; i < TABLE_COUNT; i++ {
+		cm.tables[i] = &table{
+			m: make(map[string]interface{}),
+		}
+	}
+	return cm
+}
+
+// fnv32 hash function
+func fnv32(key string) uint32 {
+	hash := uint32(2166136261)
+	for i := 0; i < len(key); i++ {
+		hash *= PRIME_32
+		hash ^= uint32(key[i])
+	}
+	return hash
+}
+
+// getTable returns the table of the given key
+func (cm *concurrentMap) getTable(key string) *table {
+	return cm.tables[fnv32(key)%uint32(cm.tableCount)]
+}
+
+// set 并发安全地修改 db 的 key -> value
+func (cm *concurrentMap) set(key string, value interface{}) int {
+	table := cm.getTable(key)
+	table.mutex.Lock()
+	defer table.mutex.Unlock()
+	_, ok := table.m[key]
+	if ok {
+		table.m[key] = value
+		return 0
+	} else {
+		table.m[key] = value
+		cm.memberCount++
+		return 1
+	}
+}
+
+// get 并发安全地获取指定 key 的 value
+func (cm *concurrentMap) get(key string) (interface{}, bool) {
+	table := cm.getTable(key)
+	table.mutex.RLock()
+	defer table.mutex.RUnlock()
+	value, ok := table.m[key]
+	if !ok {
+		return nil, false
+	} else {
+		return value, true
+	}
+}
+
+// isExisted 并发安全地判断指定 key 是否存在
+func (cm *concurrentMap) isExisted(key string) bool {
+	table := cm.getTable(key)
+	table.mutex.RLock()
+	defer table.mutex.RUnlock()
+	_, ok := table.m[key]
+	return ok
+}
+
+// delete 并发安全地删除指定 key
+func (cm *concurrentMap) delete(key string) int {
+	table := cm.getTable(key)
+	table.mutex.Lock()
+	defer table.mutex.Unlock()
+	_, ok := table.m[key]
+	if ok {
+		delete(table.m, key)
+		cm.memberCount--
+		return 1
+	} else {
+		return 0
+	}
+}

database/constants.go

@@ -0,0 +1,6 @@
+package database
+
+const (
+	INSUFFICIENT_PARAMETERS = "wrong number of arguments"
+	REQUIRE_INTEGER         = "require_integer"
+)

database/database.go

@@ -2,36 +2,61 @@ package database
 
 import (
 	"errors"
-	"sync"
+	"fmt"
+	"redis-based-on-go/redis/protocol"
 )
 
-// TODO 完善内存数据库与数据结构
-
 type DB struct {
-	table sync.Map
+	table *concurrentMap // key -> value  value 为任意 reids 类型
 }
 
 func NewDB() *DB {
-	return &DB{}
+	return &DB{table: makeConcurrentMap()}
 }
 
-func (db *DB) Set(key, value string) int {
-	_, ok := db.table.Load(key)
+// setData 修改 db 的 key -> value value 为任意 redis 类型  内部修改由各结构本身实现
+func (db *DB) setData(key string, value interface{}) int {
+	_, ok := db.table.get(key)
 	if ok {
-		db.table.Delete(key)
-		db.table.Store(key, value)
+		db.table.delete(key)
+		db.table.set(key, value)
 		return 0
 	} else {
-		db.table.Store(key, value)
+		db.table.set(key, value)
 		return 1
 	}
 }
 
-func (db *DB) Get(key string) (string, error) {
-	value, ok := db.table.Load(key)
+// getData 获取 指定 key 的 value
+func (db *DB) getData(key string) (interface{}, error) {
+	value, ok := db.table.get(key)
+	if !ok {
+		return nil, errors.New(fmt.Sprintf("key '%s' not found", key))
+	} else {
+		return value, nil
+	}
+}
+
+// isExisted 判断指定 key 是否存在
+func (db *DB) isExisted(key string) bool {
+	_, ok := db.table.get(key)
+	return ok
+}
+
+func InitDBCommand() {
+	initStringCommand()
+	initHashCommand()
+	initSetCommand()
+	initListCommand()
+	initSortedSetCommand()
+}
+
+// Execute 根据命令字段执行相应命令
+func (db *DB) Execute(args [][]byte) protocol.Reply {
+	command, ok := commandTable[string(args[0])]
 	if !ok {
-		return "", errors.New("not found")
+		return &protocol.Error{Data: fmt.Sprintf("command '%s' not found", args[0])}
 	} else {
-		return value.(string), nil
+		return command.executor(db, args[1:])
 	}
 }

database/hash_command.go

@@ -0,0 +1,87 @@
+package database
+
+import (
+	"fmt"
+	"redis-based-on-go/redis/protocol"
+)
+
+// HSET KEY_NAME FIELD VALUE
+// setHash 为哈希表中的字段赋值  若哈希表不存在则创建并进行 HSET 操作  若字段已经存在于哈希表中则覆盖
+func setHash(db *DB, args [][]byte) protocol.Reply {
+	if len(args) < 3 || len(args)%2 != 1 {
+		return &protocol.ARGUEMENTS_NUMBER_ERR_REPLY
+	}
+	var hash *Hash
+	existed := db.isExisted(string(args[0]))
+	if existed {
+		data, err := db.getData(string(args[0]))
+		if err != nil {
+			return &protocol.Error{Data: err.Error()}
+		}
+		var ok bool
+		hash, ok = data.(*Hash)
+		if !ok {
+			return &protocol.TYPE_ERR_REPLY
+		}
+	} else {
+		hash = makeSimpleHash()
+	}
+	insertCount := 0
+	for i := 0; i < len(args)/2; i++ {
+		insertCount += hash.set(string(args[2*i+1]), string(args[2*i+2]))
+	}
+	db.setData(string(args[0]), hash)
+	return &protocol.Integer{Data: insertCount}
+}
+
+// HGET KEY_NAME FIELD_NAME
+// getHash 返回哈希表中指定字段的值
+func getHash(db *DB, args [][]byte) protocol.Reply {
+	if len(args) < 2 {
+		return &protocol.ARGUEMENTS_NUMBER_ERR_REPLY
+	}
+	data, err := db.getData(string(args[0]))
+	if err != nil {
+		return &protocol.Error{Data: err.Error()}
+	} else {
+		hash, ok := data.(*Hash)
+		if !ok {
+			return &protocol.TYPE_ERR_REPLY
+		}
+		value, ok := hash.table[string(args[1])]
+		if !ok {
+			return &protocol.Error{Data: fmt.Sprintf("hash field '%s' not found", args[1])}
+		} else {
+			return &protocol.BulkString{Data: value}
+		}
+	}
+}
+
+// HGETALL KEY_NAME
+// getAllField 返回哈希表中所有的字段和值  若 key 不存在则返回空列表
+func getAllField(db *DB, args [][]byte) protocol.Reply {
+	if len(args) != 1 {
+		return &protocol.ARGUEMENTS_NUMBER_ERR_REPLY
+	}
+	data, err := db.getData(string(args[0]))
+	if err != nil {
+		return &protocol.EMPTY_ARRAY
+	} else {
+		hash, ok := data.(*Hash)
+		if !ok {
+			return &protocol.TYPE_ERR_REPLY
+		}
+		all := hash.getAll()
+		replys := make([]protocol.Reply, len(all))
+		for i, s := range all {
+			replys[i] = &protocol.BulkString{Data: s}
+		}
+		return &protocol.Array{Data: replys}
+	}
+}
+
+func initHashCommand() {
+	commandTable["hset"] = &command{name: "hset", executor: setHash}
+	commandTable["hget"] = &command{name: "hget", executor: getHash}
+	commandTable["hgetall"] = &command{name: "hgetall", executor: getAllField}
+}

database/hash_struct.go

@@ -0,0 +1,54 @@
+package database
+
+type Hash struct {
+	table map[string]string
+}
+
+// HashConsumer 遍历函数
+type HashConsumer func(key, value string) bool
+
+// MakeSimple makes a new hash table
+func makeSimpleHash() *Hash {
+	return &Hash{
+		table: make(map[string]string),
+	}
+}
+
+func (sh *Hash) get(field string) (string, bool) {
+	value, ok := sh.table[field]
+	return value, ok
+}
+
+// 0 update  1 insert
+func (sh *Hash) set(field, value string) int {
+	_, ok := sh.table[field]
+	sh.table[field] = value
+	if ok {
+		return 0
+	} else {
+		return 1
+	}
+}
+
+func (sh *Hash) size() int {
+	size := len(sh.table)
+	return size
+}
+
+func (sh *Hash) getAll() []string {
+	all := make([]string, sh.size()*2)
+	i := 0
+	sh.forEach(func(key, value string) bool {
+		all[i] = key
+		all[i+1] = value
+		i += 2
+		return true
+	})
+	return all
+}
+
+func (sh *Hash) forEach(consumer HashConsumer) {
+	for key, value := range sh.table {
+		consumer(key, value)
+	}
+}