一个非常不错的 Go 练手项目

Go编程时光

共 17901字,需浏览 36分钟

 ·

2021-06-25 00:46

作者:HDT3213


今天给大家带来的开源项目是 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([]byte4 + 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 服务器就可以开始工作啦。

最后,以上代码均简化自我写的 Godis:一个开源仅用 Go 语言实现的 Redis 服务器。期待您的关注和 Star:

项目地址:https://github.com/HDT3213/godis

四、结束

很多朋友的日常工作主要是编写业务代码,对于框架、数据库、中间件这些“架构”、“底层代码” 有一些恐惧感。

但本文我们只写了 3 个组件,共计几百行代码就实现了一个基本的 Redis 服务器。所以底层的技术并不难,只要你对技术感兴趣由浅入深、从简到繁,“底层代码”也并不神秘。


   


喜欢明哥文章的同学
欢迎长按下图订阅!

⬇⬇⬇

浏览 81
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报
评论
图片
表情
推荐
点赞
评论
收藏
分享

手机扫一扫分享

分享
举报