徒手用 Go 写个 Redis 服务器(Godis)

火币

火币HTX

火币是全球三大交易所之一,虚拟货币交易平台安全可靠,注册领取新人礼包!

图片[1]-玩转网徒手用 Go 写个 Redis 服务器(Godis)-玩转网

今天给大家带来的开源项目是 Godis:一个用 Go 语言实现的 Redis 服务器。支持:

  • 5 种数据结构(string、list、hash、set、sortedset)
  • 自动过期(TTL)
  • 发布订阅、地理位置、持久化等功能

图片[2]-玩转网徒手用 Go 写个 Redis 服务器(Godis)-玩转网

你或许不需要自己实现 Redis 服务,但你是否厌烦了每天都是写增删改查的业务代码,想提高编程水平试图从零写个项目打开 IDE 却发现无从下手?

动手造轮子一定是提高编程能力的好办法,下面就带大家用 Go 从头开始写一个 Redis 服务器(Godis),从中你将学到:

  • 如何编写 Go 语言 TCP 服务器
  • 设计并实现安全可靠的通信协议(redis 协议)
  • 如何使用 Go 语言开发高并发程序
  • 设计和实现分布式集群以及分布式事务
  • 熟悉链表、哈希表、跳表以及时间轮等常用数据结构

千万不要担心内容太难,学不会或者没有 Go 语言基础!!虽然示例代码是 Go 但不会影响你理解 Redis 的原理和底层协议以及高性能的秘密。而且作者为了照顾到广大读者,对技术的讲解做了优化。示例代码在原项目基础上做了简化,并逐行地加了注释。如果是高级玩家,请直接访问项目阅读源码:

https://github.com/HDT3213/godis

下面正文开始,让我们一起拨开 Redis 的迷雾。

图片[3]-玩转网徒手用 Go 写个 Redis 服务器(Godis)-玩转网

一、写个 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") 
} 

:ok_hand: 至此只用了 40 行代码就搞定服务端啦!启动上面的 TCP 服务后,在终端中输入 telnet 127.0.0.1 8000 就可以连接到刚写好的服务器,它会将你发送的消息原样返回给你(所以请不要骂它):

图片[4]-玩转网徒手用 Go 写个 Redis 服务器(Godis)-玩转网

这个 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 高性能的:secret:密,可以搜索 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 通过第一个字符来表示格式:

$ 
* 

下面让我们通过一些实际例子来理解协议。

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 报文:

  1. 3 
  2. $3 
  3. SET 
  4. $3 
  5. key 
  6. $5 
  7. 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 指令,这条指令的正确报文是这样的:

  1. *3   
  2. $3 
  3. SET 
  4. $4 
  5. a\r\nb  
  6. $11 
  7. hellogithub 

当 ReadBytes 读取到第五行 “a\r\nb\r\n” 时会将其误认为两行:

  1. *3   
  2. $3 
  3. SET 
  4. $4 
  5. a  // 错误的分行 
  6. // 错误的分行 
  7. $11 
  8. hellogithub 

因此当读取到第四行 $4 后,不应该继续使用 ReadBytes('\n') 读取下一行,应使用 io.ReadFull(reader, msg) 方法来读取指定长度的内容。

  1. msg = make([]byte4 + 2// 正文长度4 + 换行符长度2 
  2. _, err = io.ReadFull(reader, msg) 

2.5 编写 RESP 协议解析器

解决完上面内容包含 “\r\n” 的问题,我们就可以开始放手编写 Redis 协议解析器啦!

  1. type Payload struct { 
  2.     Data redis.Reply 
  3.     Err  error 
  4.  
  5. // ParseStream 通过 io.Reader 读取数据并将结果通过 channel 将结果返回给调用者 
  6. // 流式处理的接口适合供客户端/服务端使用 
  7. func ParseStream(reader io.Reader) <-chan *Payload { 
  8.     ch := make(chan *Payload) 
  9.     go parse0(reader, ch) 
  10.     return ch 

由于解析器的代码比较多,这里只简单地介绍一下核心流程。

  1. func parse0(reader io.Reader, ch chan<- *Payload) { 
  2.     // 初始化读取状态 
  3.     readingMultiLine := false 
  4.     expectedArgsCount := 0 
  5.     var args [][]byte 
  6.     var bulkLen int64 
  7.     for { 
  8.         // 上文中我们提到 RESP 是以行为单位的 
  9.         // 因为行分为简单字符串和二进制安全的 BulkString,我们需要封装一个 readLine 函数来兼容 
  10.         line, err = readLine(reader, bulkLen) 
  11.         if err != nil {  
  12.             // 处理错误 
  13.             return 
  14.         } 
  15.         // 接下来我们对刚刚读取的行进行解析 
  16.         // 我们简单的将 Reply 分为两类: 
  17.         // 单行: StatusReply, IntReply, ErrorReply 
  18.         // 多行: BulkReply, MultiBulkReply 
  19.  
  20.         if !readingMultiLine { 
  21.             if isMulitBulkHeader(line) { 
  22.                 // 我们收到了 MulitBulkReply 的第一行 
  23.                 // 获得 MulitBulkReply 中 BulkString 的个数 
  24.                 expectedArgsCount = parseMulitBulkHeader(line) 
  25.                 // 等待 MulitBulkReply 后续行 
  26.                 readingMultiLine = true 
  27.             } else if isBulkHeader(line) { 
  28.                 // 我们收到了 BulkReply 的第一行 
  29.                 // 获得 BulkReply 第二行的长度, 通过 bulkLen 告诉 readLine 函数下一行 BulkString 的长度 
  30.                 bulkLen = parseBulkHeader() 
  31.                 // 这个 Reply 中一共有 1 个 BulkString 
  32.                 expectedArgsCount = 1  
  33.                 // 等待 BulkReply 后续行 
  34.                 readingMultiLine = true 
  35.             } else { 
  36.                 // 处理 StatusReply, IntReply, ErrorReply 等单行 Reply 
  37.                 reply := parseSingleLineReply(line) 
  38.                 // 通过 ch 返回结果 
  39.                 emitReply(ch) 
  40.             } 
  41.         } else { 
  42.             // 进入此分支说明我们正在等待 MulitBulkReply 或 BulkReply 的后续行 
  43.             // MulitBulkReply 的后续行有两种,BulkHeader 或者 BulkString 
  44.             if isBulkHeader(line) { 
  45.                 bulkLen = parseBulkHeader() 
  46.             } else { 
  47.                 // 我们正在读取一个 BulkString, 它可能是 MulitBulkReply 或 BulkReply  
  48.                 args = append(args, line) 
  49.             } 
  50.             if len(args) == expectedArgsCount { // 我们已经读取了所有后续行 
  51.                 // 通过 ch 返回结果 
  52.                 emitReply(ch) 
  53.                 // 重置状态, 准备解析下一条 Reply 
  54.                 readingMultiLine = false 
  55.                 expectedArgsCount = 0 
  56.                 args = nil 
  57.                 bulkLen = 0 
  58.             } 
  59.         } 
  60.     } 

三、实现内存数据库

至此我们已经搞定数据接收和解析的部分了,剩下就是我们应该把数据存在哪里了?

抛开持久化部分,作为基于内存的 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 造成影响。

图片[5]-玩转网徒手用 Go 写个 Redis 服务器(Godis)-玩转网

代码如下:

  1. type ConcurrentDict struct { 
  2.     table []*Shard 
  3.     count int32 
  4.  
  5. type Shard struct { 
  6.     m     map[string]interface{} 
  7.     mutex sync.RWMutex 
  8.  
  9. func (dict *ConcurrentDict) spread(hashCode uint32) uint32 { 
  10.     tableSize := uint32(len(dict.table)) 
  11.     return (tableSize – 1) & uint32(hashCode) 
  12.  
  13. func (dict *ConcurrentDict) getShard(index uint32) *Shard { 
  14.     return dict.table[index] 
  15.  
  16. func (dict *ConcurrentDict) Get(key string) (val interface{}, exists bool) { 
  17.     hashCode := fnv32(key) 
  18.     index := dict.spread(hashCode) 
  19.     shard := dict.getShard(index) 
  20.     shard.mutex.RLock() 
  21.     defer shard.mutex.RUnlock() 
  22.     val, exists = shard.m[key] 
  23.     return 
  24.  
  25. func (dict *ConcurrentDict) Put(key string, val interface{}) (result int) { 
  26.     if dict == nil { 
  27.         panic(“dict is nil”
  28.     } 
  29.     hashCode := fnv32(key) 
  30.     index := dict.spread(hashCode) 
  31.     shard := dict.getShard(index) 
  32.     shard.mutex.Lock() 
  33.     defer shard.mutex.Unlock() 
  34.  
  35.     if _, ok := shard.m[key]; ok { 
  36.         shard.m[key] = val 
  37.         return 0 
  38.     } else { 
  39.         shard.m[key] = val 
  40.         dict.addCount() 
  41.         return 1 
  42.     } 

ConcurrentDict 可以保证对单个 key 操作的并发安全性,但是仍然无法满足并发安全的需求,举例来说:

  1. 读取 -> 做加法 -> 写入 

因此我们需要实现 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 所在的哈希槽也可以保证安全,另一方面哈希槽数量较少即使不释放也不会消耗太多内存。

  1. type Locks struct { 
  2.     table []*sync.RWMutex 
  3.  
  4. func Make(tableSize int) *Locks { 
  5.     table := make([]*sync.RWMutex, tableSize) 
  6.     for i := 0; i < tableSize; i++ { 
  7.         table[i] = &sync.RWMutex{} 
  8.     } 
  9.     return &Locks{ 
  10.         table: table, 
  11.     } 
  12.  
  13. func (locks *Locks)Lock(key string) { 
  14.     index := locks.spread(fnv32(key)) 
  15.     mu := locks.table[index] 
  16.     mu.Lock() 
  17.  
  18. func (locks *Locks)UnLock(key string) { 
  19.     index := locks.spread(fnv32(key)) 
  20.     mu := locks.table[index] 
  21.     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 服务器。所以底层的技术并不难,只要你对技术感兴趣由浅入深、从简到繁,“底层代码”也并不神秘。

© 版权声明
THE END
点赞22赞赏 分享