一个非常不错的 Go 练手项目
今天给大家带来的开源项目是 Godis:一个用 Go 语言实现的 Redis 服务器。支持:
5 种数据结构(string、list、hash、set、sortedset) 自动过期(TTL) 发布订阅、地理位置、持久化等功能
你或许不需要自己实现 Redis 服务,但你是否厌烦了每天都是写增删改查的业务代码,想提高编程水平试图从零写个项目打开 IDE 却发现无从下手?
动手造轮子一定是提高编程能力的好办法,下面就带大家用 Go 从零开始写一个 Redis 服务器(Godis),从中你将学到:
如何编写 Go 语言 TCP 服务器 设计并实现安全可靠的通信协议(redis 协议) 如何使用 Go 语言开发高并发程序 设计和实现分布式集群以及分布式事务 熟悉链表、哈希表、跳表以及时间轮等常用数据结构
千万不要担心内容太难!!虽然示例代码是 Go,但就算你不会 Go 语言也不会影响你理解 Redis 的原理和底层协议以及高性能的秘密。而且作者为了照顾到广大读者,对技术的讲解做了优化。示例代码在原项目基础上做了简化,并逐行地加了注释。如果是高级玩家,请直接访问项目阅读源码:
https://github.com/HDT3213/godis
下面让我们一起拨开 Redis 的迷雾。
一、写个 TCP 服务器
众所周知 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
,优雅关闭步骤:
首先,关闭 listener 停止接受新连接 然后,遍历所有存活连接逐个关闭
优雅关闭的代码比较多,这里就不完整贴出了。
二、透视 Redis 协议
在解决完通信后,下一步就是搞清楚 Redis 的协议,其实就是一套序列化协议类似 JSON、Protocol Buffers,你看底层其实也就是一些基础的知识。
自 Redis 2.0 以后的通信统一为 RESP 协议(REdis Serialization Protocol),该协议易于实现不仅可以高效的被程序解析,还能够被人类读懂容易调试。
RESP 是一个二进制安全的文本协议,工作于 TCP 协议上。RESP 以行作为单位,客户端和服务器发送的命令或数据一律以 \r\n
(CRLF)作为换行符。
二进制安全是指允许协议中出现任意字符而不会导致故障。比如 C 语言的字符串以 \0
作为结尾不允许字符串中间出现 \0
,而 Go 语言的 string 则允许出现 \0
,我们说 Go 语言的 string 是二进制安全的,而 C 语言字符串不是二进制安全的。
RESP 的二进制安全性允许我们在 key 或者 value 中包含 \r
或者 \n
这样的特殊字符。在使用 Redis 存储 protobuf、msgpack 等二进制数据时,二进制安全性尤为重要。
RESP 定义了 5 种格式:
简单字符串(Simple String):服务器用来返回简单的结果,比如 "OK" 非二进制安全,且不允许换行 错误信息(Error):服务器用来返回简单的错误信息,比如 "ERR Invalid Synatx" 非二进制安全,且不允许换行 整数(Integer):llen、scard 等命令的返回值,64 位有符号整数 字符串(Bulk String):二进制安全字符串,比如 get 等命令的返回值 数组(Array,又称 Multi Bulk Strings):Bulk String 数组,客户端发送指令以及 lrange 等命令响应的格式
RESP 通过第一个字符来表示格式:
简单字符串:以"+" 开始, 如:"+OK\r\n" 错误:以"-" 开始,如:"-ERR Invalid Synatx\r\n" 整数:以":"开始,如:":1\r\n" 字符串:以 $
开始数组:以 *
开始
下面让我们通过一些实际例子来理解协议。
2.1 字符串
字符串(Bulk String)有两行,第一行为 $
+正文长度,第二行为实际内容。如:
$3\r\nSET\r\n
字符串(Bulk String)是二进制安全的,就是说可以在 Bulk String 内部包含 "\r\n" 字符(行尾的 CRLF 被隐藏):
$4
a\r\nb
2.2 空
$-1
表示 nil,比如使用 get 命令查询一个不存在的 key 时,响应即为 $-1
。
2.3 数组
数组(Array)格式第一行为 "*"+数组长度,其后是相应数量的 字符串(Bulk String)。比如 ["foo", "bar"]
的报文(传输时的内容):
*2
$3
foo
$3
bar
客户端也使用 数组(Array)格式向服务端发送指令。命令本身将作为第一个参数,比如 SET key value
指令的 RESP 报文:
*3
$3
SET
$3
key
$5
value
将换行符打印出来:
*3\r\n$3\r\nSET\r\n$3\r\nkey\r\n$5\r\nvalue\r\n
2.4 解析预备
知道常用的 RESP 报文内容后,就可以开始着手解析了。但需要注意的是 RESP 是 二进制安全
的协议,它允许在正文中使用 \r\n
字符。举例来说 Redis 可以正确接收并执行 SET "a\r\nb" hellogithub
指令,这条指令的正确报文是这样的:
*3
$3
SET
$4
a\r\nb
$11
hellogithub
当 ReadBytes
读取到第五行 "a\r\nb\r\n" 时会将其误认为两行:
*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)
2.5 编写 RESP 协议解析器
解决完上面内容包含 "\r\n" 的问题,我们就可以开始放手编写 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)是并行工作的,所以我们必须考虑各种并发安全问题。常见的并发安全哈希表设计有几种:
sync.map
:Golang 官方提供的并发哈希表,适合读多写少的场景。但是在m.dirty
刚被提升后会将m.read
复制到新的m.dirty
中,在数据量较大的情况下复制操作会阻塞所有协程,存在较大的隐患。juc.ConcurrentHashMap
:Java 的并发哈希表采用分段锁实现。在进行扩容时访问哈希表线程都将协助进行 rehash 操作,在 rehash 结束前所有的读写操作都会阻塞。因为缓存数据库中键值对数量巨大且对读写操作响应时间要求较高,使用 juc 的策略是不合适的。memcached hashtable
:在后台线程进行 rehash 操作时,主线程会判断要访问的哈希槽是否已被 rehash 从而决定操作 old_hashtable 还是操作 new_hashtable。这种设计被称为渐进式 rehash 它的优点是 rehash 操作基本不会阻塞主线程的读写,是最理想的的方案。
但渐进式 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 操作的并发安全性,但是仍然无法满足并发安全的需求,举例来说:
Incr 命令需要完成: 读取 -> 做加法 -> 写入
三步操作,读取和写入两步操作不是原子性的MSETNX 命令当且仅当所有给定键都不存在时所有给定键设置值,我们需要保证「检查多个key是否存在」以及「写入多个key」这两个操作的原子性
因此我们需要实现 db.Locker
用于锁定一个或一组 key 直到我们完成所有操作后再释放。
实现 db.Locker
最直接的想法是使用一个 map[string]*sync.RWMutex
加锁过程分为两步:初始化 mutex -> 加锁 解锁过程也分为两步: 解锁 -> 释放mutex
那么存在一个无法解决的并发问题:
时间 | 协程A | 协程B |
---|---|---|
1 | locker["a"].Unlock() | |
2 | locker["a"] = &sync.RWMutex{} | |
3 | delete(locker["a"]) | |
4 | locker["a"].Lock() |
由于 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 服务器。所以底层的技术并不难,只要你对技术感兴趣由浅入深、从简到繁,“底层代码”也并不神秘。
⬇⬇⬇