作者：HDT3213
今天给大家带来的开源项目是 Godis：一个用 Go 语言实现的 redis 服务器。支持：
你或许不需要自己实现 Redis 服务，但你是否厌烦了每天都是写增删改查的业务代码，想提高编程水平试图从零写个项目打开 IDE 却发现无从下手？
动手造轮子一定是提高编程能力的好办法，下面就带大家用 Go 从零开始写一个 Redis 服务器（Godis），从中你将学到：
千万不要担心内容太难，学不会或者没有 Go 语言基础！！虽然示例代码是 Go 但不会影响你理解 Redis 的原理和底层协议以及高性能的秘密。而且作者为了照顾到广大读者，对技术的讲解做了优化。示例代码在原项目基础上做了简化，并逐行地加了注释。如果是高级玩家，请直接访问项目阅读源码：
https://github.com/HDT3213/godis
下面让我们一起拨开 Redis 的迷雾。
众所周知 Redis 是 C/S 模型，使用 TCP 协议进行通信。接下来就从实现 TCP 服务端开始。作为广泛用于服务端的编程语言 Golang 提供了非常简洁的 TCP 接口，所以实现起来十分方便。示例代码：
func ListenAndServe(address string) {
// 绑定监听地址
listener, err := net.Listen("tcp", address)
if err != nil {
log.Fatal(fmt.Sprintf("listen err: %v", err))
}
defer listener.Close()
log.Println(fmt.Sprintf("bind: %s, start listening...", address))
for {
// Accept 会一直阻塞直到有新的连接建立或者listen中断才会返回
conn, err := listener.Accept()
if err != nil {
// 通常是由于listener被关闭无法继续监听导致的错误
log.Fatal(fmt.Sprintf("accept err: %v", err))
}
// 开启新的 goroutine 处理该连接
go Handle(conn)
}
}
func Handle(conn net.Conn) {
reader := bufio.NewReader(conn)
for {
// ReadString 会一直阻塞直到遇到分隔符 'n'
// 遇到分隔符后 ReadString 会返回上次遇到分隔符到现在收到的所有数据
// 若在遇到分隔符之前发生异常, ReadString 会返回已收到的数据和错误信息
msg, err := reader.ReadString('n')
if err != nil {
// 通常遇到的错误是连接中断或被关闭，用io.EOF表示
if err == io.EOF {
log.Println("connection close")
} else {
log.Println(err)
}
return
}
b := []byte(msg)
// 将收到的信息发送给客户端
conn.Write(b)
}
}
func main() {
ListenAndServe(":8000")
}
至此只用了 40 行代码就搞定服务端啦！启动上面的 TCP 服务后，在终端中输入 telnet 127.0.0.1 8000 就可以连接到刚写好的服务器，它会将你发送的消息原样返回给你（所以请不要骂它）：
这个 TCP 服务器的非常简单，主协程调用 accept 函数来监听端口，接受新连接后开启一个 Goroutine 来处理它。这种简单的阻塞 IO 模型有些类似于早期的 Tomcat/Apache 服务器。
阻塞 IO 模型是使用一个线程处理一个连接，在没有收到新数据时监听线程处于阻塞状态，直到数据就绪后线程被唤醒进行处理。因为阻塞 IO 模型需要开启大量线程并且频繁地进行上下文切换，所以它的效率很低。而 Redis 使用的 epoll 技术（IO 多路复用）用一个线程处理大量连接，极大地提高了吞吐量。那么我们的 TCP 服务器会比 Redis 慢很多吗？
当然不会，Golang 利用 Goroutine 调度开销远远小于线程调度开销的优势封装出 goroutine-per-connection 风格的极简接口，而且 net/tcp 库将 epoll 封装成了阻塞 IO 的样子，在享受 epoll 高性能的同时避免了原生 epoll 接口所需的复杂异步代码。
在作者的电脑上 Redis 每秒可以响应 10.6k 个 PING 命令，而 Godis（完整代码） 的吞吐量为 9.2 kqps 相差并不大。想了解更多 Golang 高性能的㊙️密，可以搜索 go netpoller 或者 go 语言 网络轮询器 关键字
另外，合格的 TCP 的服务器在关闭的时候不应该一停了之，而需要完成响应已接收的请求、释放 TCP 连接等必要的清理工作。这个功能我们一般称为 优雅关闭 或者 graceful shutdown，优雅关闭步骤：
优雅关闭的代码比较多，这里就不完整贴出了。
在解决完通信后，下一步就是搞清楚 Redis 的协议，其实就是一套序列化协议类似 JSON、Protocol Buffers，你看底层其实也就是一些基础的知识。
自 Redis 2.0 以后的通信统一为 RESP 协议（REdis Serialization Protocol)，该协议易于实现不仅可以高效的被程序解析，还能够被人类读懂容易调试。
RESP 是一个二进制安全的文本协议，工作于 TCP 协议上。RESP 以行作为单位，客户端和服务器发送的命令或数据一律以 rn（CRLF）作为换行符。
二进制安全是指允许协议中出现任意字符而不会导致故障。比如 C 语言的字符串以 作为结尾不允许字符串中间出现 ，而 Go 语言的 string 则允许出现 ，我们说 Go 语言的 string 是二进制安全的，而 C 语言字符串不是二进制安全的。
RESP 的二进制安全性允许我们在 key 或者 value 中包含 r 或者 n 这样的特殊字符。在使用 Redis 存储 protobuf、msgpack 等二进制数据时，二进制安全性尤为重要。
RESP 定义了 5 种格式：
RESP 通过第一个字符来表示格式：
下面让我们通过一些实际例子来理解协议。
字符串（Bulk String）有两行，第一行为 $+正文长度，第二行为实际内容。如：
$3rnSETrn
字符串（Bulk String）是二进制安全的，就是说可以在 Bulk String 内部包含 "rn" 字符（行尾的 CRLF 被隐藏）：
$4
arnb
$-1 表示 nil，比如使用 get 命令查询一个不存在的 key 时，响应即为 $-1。
数组（Array）格式第一行为 "*"+数组长度，其后是相应数量的 字符串（Bulk String）。比如 ["foo", "bar"] 的报文（传输时的内容）：
*2
$3
foo
$3
bar
客户端也使用 数组（Array）格式向服务端发送指令。命令本身将作为第一个参数，比如 SET key value 指令的 RESP 报文：
*3
$3
SET
$3
key
$5
value
将换行符打印出来：
*3rn$3rnSETrn$3rnkeyrn$5rnvaluern
知道常用的 RESP 报文内容后，就可以开始着手解析了。但需要注意的是 RESP 是 二进制安全 的协议，它允许在正文中使用 rn 字符。举例来说 Redis 可以正确接收并执行 SET "arnb" hellogithub 指令，这条指令的正确报文是这样的：
*3
$3
SET
$4
arnb
$11
hellogithub
当 ReadBytes 读取到第五行 "arnbrn" 时会将其误认为两行：
*3
$3
SET
$4
a // 错误的分行
b // 错误的分行
$11
hellogithub
因此当读取到第四行 $4 后，不应该继续使用 ReadBytes('n') 读取下一行，应使用 io.ReadFull(reader, msg) 方法来读取指定长度的内容。
msg = make([]byte, 4 + 2) // 正文长度4 + 换行符长度2
_, err = io.ReadFull(reader, msg)
解决完上面内容包含 "rn" 的问题，我们就可以开始放手编写 Redis 协议解析器啦！
type Payload struct {
Data redis.Reply
Err error
}
// ParseStream 通过 io.Reader 读取数据并将结果通过 channel 将结果返回给调用者
// 流式处理的接口适合供客户端/服务端使用
func ParseStream(reader io.Reader) <-chan *Payload {
ch := make(chan *Payload)
go parse0(reader, ch)
return ch
}
由于解析器的代码比较多，这里只简单地介绍一下核心流程。
func parse0(reader io.Reader, ch chan<- *Payload) {
// 初始化读取状态
readingMultiLine := false
expectedArgsCount := 0
var args [][]byte
var bulkLen int64
for {
// 上文中我们提到 RESP 是以行为单位的
// 因为行分为简单字符串和二进制安全的 BulkString，我们需要封装一个 readLine 函数来兼容
line, err = readLine(reader, bulkLen)
if err != nil {
// 处理错误
return
}
// 接下来我们对刚刚读取的行进行解析
// 我们简单的将 Reply 分为两类:
// 单行: StatusReply, IntReply, ErrorReply
// 多行: BulkReply, MultiBulkReply
if !readingMultiLine {
if isMulitBulkHeader(line) {
// 我们收到了 MulitBulkReply 的第一行
// 获得 MulitBulkReply 中 BulkString 的个数
expectedArgsCount = parseMulitBulkHeader(line)
// 等待 MulitBulkReply 后续行
readingMultiLine = true
} else if isBulkHeader(line) {
// 我们收到了 BulkReply 的第一行
// 获得 BulkReply 第二行的长度, 通过 bulkLen 告诉 readLine 函数下一行 BulkString 的长度
bulkLen = parseBulkHeader()
// 这个 Reply 中一共有 1 个 BulkString
expectedArgsCount = 1
// 等待 BulkReply 后续行
readingMultiLine = true
} else {
// 处理 StatusReply, IntReply, ErrorReply 等单行 Reply
reply := parseSingleLineReply(line)
// 通过 ch 返回结果
emitReply(ch)
}
} else {
// 进入此分支说明我们正在等待 MulitBulkReply 或 BulkReply 的后续行
// MulitBulkReply 的后续行有两种，BulkHeader 或者 BulkString
if isBulkHeader(line) {
bulkLen = parseBulkHeader()
} else {
// 我们正在读取一个 BulkString, 它可能是 MulitBulkReply 或 BulkReply
args = Append(args, line)
}
if len(args) == expectedArgsCount { // 我们已经读取了所有后续行
// 通过 ch 返回结果
emitReply(ch)
// 重置状态, 准备解析下一条 Reply
readingMultiLine = false
expectedArgsCount = 0
args = nil
bulkLen = 0
}
}
}
}
至此我们已经搞定数据接收和解析的部分了，剩下就是我们应该把数据存在哪里了？
抛开持久化部分，作为基于内存的 KV 数据库 Redis 的所有数据需要都存储在内存中的哈希表，而这个哈希表就是我们今天需要编写的最后一个组件。
与单线程的 Redis 不同我们实现的 Redis（godis）是并行工作的，所以我们必须考虑各种并发安全问题。常见的并发安全哈希表设计有几种：
但渐进式 rehash 的实现非常复杂，所以 godis 采用 Golang 社区广泛使用的分段锁策略（非上面的三种），就是将 key 分散到固定数量的 shard 中避免进行整体 rehash 操作。shard 是有锁保护的 map，当 shard 进行 rehash 时会阻塞 shard 内的读写，但不会对其他 shard 造成影响。
代码如下：
type ConcurrentDict struct {
table []*Shard
count int32
}
type Shard struct {
m map[string]interface{}
mutex sync.RWMutex
}
func (dict *ConcurrentDict) spread(hashCode uint32) uint32 {
tableSize := uint32(len(dict.table))
return (tableSize - 1) & uint32(hashCode)
}
func (dict *ConcurrentDict) getShard(index uint32) *Shard {
return dict.table[index]
}
func (dict *ConcurrentDict) Get(key string) (val interface{}, exists bool) {
hashCode := fnv32(key)
index := dict.spread(hashCode)
shard := dict.getShard(index)
shard.mutex.RLock()
defer shard.mutex.RUnlock()
val, exists = shard.m[key]
return
}
func (dict *ConcurrentDict) Put(key string, val interface{}) (result int) {
if dict == nil {
panic("dict is nil")
}
hashCode := fnv32(key)
index := dict.spread(hashCode)
shard := dict.getShard(index)
shard.mutex.Lock()
defer shard.mutex.Unlock()
if _, ok := shard.m[key]; ok {
shard.m[key] = val
return 0
} else {
shard.m[key] = val
dict.addCount()
return 1
}
}
ConcurrentDict 可以保证对单个 key 操作的并发安全性，但是仍然无法满足并发安全的需求，举例来说：
因此我们需要实现 db.Locker 用于锁定一个或一组 key 直到我们完成所有操作后再释放。
实现 db.Locker 最直接的想法是使用一个 map[string]*sync.RWMutex
那么存在一个无法解决的并发问题：
由于 t3 时协程 B 释放了锁，t4 时协程 A 试图加锁会失败。若协程B在解锁时不执行 delete(locker["a"]) 就可以避免该异常的发生，但是这样会造成严重的内存泄露。
我们注意到哈希槽的数量远少于 key 的数量，反过来说多个键可以共用一个哈希槽。所以我们不再直接对 key 进行加锁而是锁定 key 所在的哈希槽也可以保证安全，另一方面哈希槽数量较少即使不释放也不会消耗太多内存。
type Locks struct {
table []*sync.RWMutex
}
func Make(tableSize int) *Locks {
table := make([]*sync.RWMutex, tableSize)
for i := 0; i < tableSize; i++ {
table[i] = &sync.RWMutex{}
}
return &Locks{
table: table,
}
}
func (locks *Locks)Lock(key string) {
index := locks.spread(fnv32(key))
mu := locks.table[index]
mu.Lock()
}
func (locks *Locks)UnLock(key string) {
index := locks.spread(fnv32(key))
mu := locks.table[index]
mu.Unlock()
}
在锁定多个 key 时需要注意，若 协程A 持有 键a 的锁试图获得 键b 的锁，此时 协程B 持有 键b 的锁试图获得 键a 的锁则会形成死锁。
解决方法是所有协程都按照相同顺序加锁，若两个协程都想获得 键a 和 键b 的锁，那么必须先获取 键a 的锁后获取 键b 的锁，这样就可以避免循环等待。
到目前为止构建 Redis 服务器所需的基本组件已经备齐，只需要将 TCP 服务器、协议解析器与哈希表组装起来我们的 Redis 服务器就可以开始工作啦。
最后，以上代码均简化自我写的 Godis：一个开源仅用 Go 语言实现的 Redis 服务器。期待您的关注和 Star：
项目地址：
https://github.com/HDT3213/godis
很多朋友的日常工作主要是编写业务代码，对于框架、数据库、中间件这些“架构”、“底层代码” 有一些恐惧感。
但本文我们只写了 3 个组件，共计几百行代码就实现了一个基本的 Redis 服务器。所以底层的技术并不难，只要你对技术感兴趣由浅入深、从简到繁，“底层代码”也并不神秘。
京公网安备 11010802035086号