Golang实现自己的Redis数据库内存实例探究

# ping服务器
PING [message]
# 授权密码设置
AUTH <password>
# 选择数据库
SELECT index
# 设置key
SET key value [NX | XX]  [EX seconds | PX milliseconds]
# 获取key
GET key

// redisCommand 待执行的命令  protocal.Reply 执行结果
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:])
	}
	 //... 省略...
}

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

	iflen(redisArgs) == 0 { // 不带参数
		return protocal.NewPONGReply()
	} elseiflen(redisArgs) == 1 { // 带参数1个
		return protocal.NewBulkReply(redisArgs[0])
	}
    // 否则，回复命令格式错误
	return protocal.NewArgNumErrReply("ping")
}

// redisCommand 待执行的命令  protocal.Reply 执行结果
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")
	}

	//... 省略...
}

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()
}

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

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

// redisCommand 待执行的命令  protocal.Reply 执行结果
func (e *Engine) Exec(c *connection.KeepConnection, redisCommand [][]byte) (result protocal.Reply) {
	//....忽略....
	// 基础命令
	switch commandName {
	case"select": // 表示当前连接，要选中哪个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()
}

// redisCommand 待执行的命令  protocal.Reply 执行结果
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)
}

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]))
	// 从命令注册中心，获取命令的执行函数
	command, ok := commandCenter[cmdName]
	if !ok {
		return protocal.NewGenericErrReply("unknown command '" + cmdName + "'")
	}
	fun := command.execFunc
	return fun(db, redisCommand[1:])
}

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

// 并发安全的字典
type ConcurrentDict struct {
	shds  []*shard      // 底层shard切片
	mask  uint32// 掩码
	count *atomic.Int32 // 元素个数
}

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

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

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)
}
// 获取底层存储对象【字节流】
func (db *DB) getStringObject(key string) ([]byte, protocal.Reply) {
	payload, exist := db.GetEntity(key)
	if !exist {
		returnnil, protocal.NewNullBulkReply()
	}
	// 判断底层对象是否为【字节流】
	bytes, ok := payload.RedisObject.([]byte)
	if !ok {
		returnnil, protocal.NewWrongTypeErrReply()
	}
	return bytes, nil
}
// 获取内存中的数据
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
}