为庆祝35岁生日,秃顶的我徒手用 Go 写个 Redis 服务器

Posted javatiange

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了为庆祝35岁生日,秃顶的我徒手用 Go 写个 Redis 服务器相关的知识,希望对你有一定的参考价值。

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

  • 5 种数据结构(string、list、hash、set、sortedset)

  • 自动过期(TTL)

  • 发布订阅、地理位置、持久化等功能

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

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

  • 如何编写 Go 语言 TCP 服务器

  • 设计并实现安全可靠的通信协议(redis 协议)

  • 如何使用 Go 语言开发高并发程序

  • 设计和实现分布式集群以及分布式事务

  • 熟悉链表、哈希表、跳表以及时间轮等常用数据结构

千万不要担心内容太难!!虽然示例代码是 Go,但就算你不会 Go 语言也不会影响你理解 Redis 的原理和底层协议以及高性能的秘密。而且作者为了照顾到广大读者,对技术的讲解做了优化。示例代码在原项目基础上做了简化,并逐行地加了注释。

下面让我们一起拨开 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 操作的并发安全性,但是仍然无法满足并发安全的需求,举例来说:

  1. Incr 命令需要完成:读取 -> 做加法 -> 写入 三步操作,读取和写入两步操作不是原子性的

  2. MSETNX 命令当且仅当所有给定键都不存在时所有给定键设置值,我们需要保证「检查多个key是否存在」以及「写入多个key」这两个操作的原子性

因此我们需要实现 db.Locker 用于锁定一个或一组 key 直到我们完成所有操作后再释放。

实现 db.Locker 最直接的想法是使用一个 map[string]*sync.RWMutex

  • 加锁过程分为两步:初始化 mutex -> 加锁

  • 解锁过程也分为两步: 解锁 -> 释放mutex

那么存在一个无法解决的并发问题:

时间协程A协程B
1
locker[“a”].Unlock()
2locker[“a”] = &sync.RWMutex{}
3
delete(locker[“a”])
4locker[“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 服务器就可以开始工作啦。

最后

内部独家教材—Redis底层原理剖析

免费为粉丝准备了超级干货内部教材~

《Redis底层原理及架构实战笔记及高频面试解析》:
在这里插入图片描述

看完你能收获如下

  • 架构级Redis底层源码设计剖析实践

  • Hotspot源码深度分析,趣图精读Redis启动流程及多路复用源码

  • 结合一线大厂(阿里、美团等)充分讲解Redis在互联网高并发环境中的实战场景

  • 全方位实战解刨Redis使用过程产生的热点问题与其解决方案

  • 真实揭秘大厂面试套路,深入底层原理分析Redis连环炮问题

需要的小伙伴,可以一键三连,下方获取免费领取方式!
在这里插入图片描述

以上是关于为庆祝35岁生日,秃顶的我徒手用 Go 写个 Redis 服务器的主要内容,如果未能解决你的问题,请参考以下文章

Debian 官方发文庆祝 28 岁生日

诸神眷顾的幻想乡

谷歌软件工程师:我为什么喜欢用Go语言?

怎样用生日算年龄

HTML5庆祝生日蛋糕烟花特效

[Zjoi2015]诸神眷顾的幻想乡