当前位置：网站首页>「Redis源码系列」关于源码阅读的学习与思考

「Redis源码系列」关于源码阅读的学习与思考

2022-07-02 06:33:00 【程序员柒柒】

前言

通过之前的源码阅读与分析, 我们通过服务的启动, 数据流的接受与处理, 整体DB结构, 详细的存储数据结构等方面的学习对于Redis6.0有了一个较为系统的认知。尤其是其中的一些优秀的设计我们在学习完成后也要深入的加以分析和思考, 是否可以将这些经验借鉴到我们实际的工作中呢？这次就和大家一起讨论一下我学习完成后的收获。

服务器模型

在Redis6.0之前的版本服务器采用单进程单线程的处理方式, 优点就是避免了并发的锁开销, 缺点是不能充分利用CPU的多核处理。在现在业务场景中, CPU通常不会成为负载的主要瓶颈, 更多在于内存和网络。Redis 的多线程网络模型实际上并不是一个标准的
Multi-Reactors/Master-Workers 模型, 在6.0.0的版本中, I/O线程只是负责完成I/O流的处理任务, 当然主线程也会承担部分I/O任务, 真正处理命令执行的在主线程中完成。这样依然可以保持无锁处理指令。这样既保持了原系统的兼容性，又能利用多核提升 I/O 性能。这给我带来的思考是我们不必要完全拘泥于历史经验的典型处理方法。针对于不同的场景可以做我们自己的服务器处理策略。

分享一个case, 在高并发的分布式系统中, 对于同一个用户请求的处理可能涉及到加锁, 事务操作, 幂等性校验等情况。类比于redis对于命令的处理, 本质上是相同的, 用户请求的数据相当于set操作, 数据库中的数据相当于RedisDB中待操作的对象, 在业务中我们一般会通过分布式锁控制并发, 使用数据库的隔离级别控制事务锁, 使用数据库唯一键做幂等控制等。这种通用性方案可以在很大程度上避免错误的发生, 但是性能上却受到一定的影响，而且处理不当会经常有死锁的发生, 那么我们是否可以参考Redis在多线程上的处理方式实现业务呢？答案是可以的。

如图, 在server中我们实现一个典型的Master-Workers架构的tcp服务器, 可以根据配置文件配置指定的线程池数量处理用户请求。

// 创建process
func (process *taskProcess) createGoPool() {
   for i := 0; i < process.maxGo; i++ {
      go func(processId int) {

         ch := make(chan *server.ServerContext, 128)
         process.goPool.Store(processId, ch)

         for {
            select {
            case data, ok := <-ch:
               if !ok {
                  fmt.Println("process is not ok ", processId)
               }
               // 处理请求数据
               process.dealData(data)
            }
         }
      }(i)
   }
}

当请求到来的时候分发规则就是我们可以自定义的, 加入我们的业务可以根据UID来进行请求划分, 则可以根据UID%threadNum指定processId来分发请求到指定的goroutine中来处理, 伪代码如下:

func (dispatcher *Dispatcher) parseMsg(info *requestInfo) {

   // 解析body数据
   requestBody := &server.RequestMsgBody{}
   err := json.Unmarshal(info.msg, &requestBody)
   if err != nil {
      writeinfo := fmt.Sprintf("解析发生错误: %+v, 消息丢弃: %s", err, string(info.msg))
      write, _ := comm.GbkToUtf8([]byte(writeinfo))
      info.conn.Write(write)
      return
   }

   // 如果存在当前用户请求的ID, 则使用同一线程处理, 否则使用空闲线程处理并记录用户ID与线程ID
   var processId int
   if val, isExistsUidProcessId := dispatcher.activeUserProcessMap.Load(requestBody.Md.Uid); buyOk {
      processId = val.(int)
   } else {
      processId = NewTaskProcess.GetIdleProcess()
   }

   // 将数据分发
   taskData := &server.ServerContext{
      Conn: info.conn,
      Body: requestBody,
   }

   // 发送数据分发消息
   ch, ok := NewTaskProcess.goPool.Load(processId)
   if ok {
      ch.(chan *server.ServerContext) <- taskData
   }

   // 维护UID对processId的关系
   // 发送成功，保存map & 计数
   if requestBody.Md.Uid > 0 {
      dispatcher.activeUserProcessMap.Store(requestBody.Md.Uid, processId)
      dispatcher.activeUserCount(requestBody.Md.Uid, 1)
   }

}

基于这样的方案可以将用户请求执行单线程的处理, 不用担心任何的锁消耗, 当然在负载均衡网关层还需要添加请求分发的插件, 将请求与集群的server进行绑定, 这样就可以模拟实现单线程处理用户的请求, 在并发量极高的时候, 可以极大避免各种锁带来的开销。并且处理能力也可以提升30%。当然这样并不是通用的处理方案, 是基于特殊的业务场景做特殊的实现。也是对我们的一种启发。

渐进式rehash

当数据库中存在上百万的key, 占用数GB内存的时候, 需要扩容就是一件非常耗时的操作, 同时CPU使用率也会飙升。对此redis使用了渐进式的扩容策略, 复习一下redishashtable的结构:


#dict字典的数据结构
typedef struct dict{
    dictType *type; //直线dictType结构，dictType结构中包含自定义的函数，这些函数使得key和value能够存储任何类型的数据
    void *privdata; //私有数据，保存着dictType结构中函数的 参数
    dictht ht[2]; //两张哈希表
    long rehashidx; //rehash的标记，rehashidx=-1表示没有进行rehash，rehash时每迁移一个桶就对rehashidx加一
    int itreators;  //正在迭代的迭代器数量
}
 
#dict结构中ht[0]、ht[1]哈希表的数据结构
typedef struct dictht{
    dictEntry[] table;        //存放一个数组的地址，数组中存放哈希节点dictEntry的地址
    unsingned long size;      //哈希表table的大小，出始大小为4
    unsingned long  sizemask; //用于将hash值映射到table位置的索引，大小为（size-1）
    unsingned long  used;     //记录哈希表已有节点（键值对）的数量
}

当达到扩容条件时, 会将rehashidx置为0标识rehash开始, 在渐进式rehash进行期间，字典的删除、查找、更新等操作会在两个哈希表上进行。比如说，要在字典里面查找一个键的话，程序会先在ht[0]里面进行查找，如果没找到的话，就会继续到ht[1]里面进行查找, 新添加到字典的键值对一律会被保存到 ht[1] 里面, 以此操作在某一个时间点, 会将ht[0]的数据迁移完成, 并且将ht[1]置为ht[0], ht[1]置为NULL, 为下一次rehash做准备。
在一次和新浪同学的交流中, 发现他们也是使用了类似的方法来更新本地缓存, 我们知道新浪曾经是最大的redis集群, 在热点新闻爆发的时候点赞, 评论, 转发，热点新闻等会量级会指数量级攀升, 如果使用Redis性能也是远远不够, 所以他们同时也使用了大量的本地缓存, 使用本地缓存就要涉及到缓存的及时更新, 在更新大量本地缓存是也同样不可避免的要涉及到读写锁开销, 大量扩容带来的CPU消耗等, 为此他们也是采用了两套map来规避此类问题, 业务模型大致如下:

当请求到来的时候会优先读取本地缓存(内存)数据, 同时本地缓存会有各种来源的设置, 当数据量比较大的时候就会有非常大的cpu抖动, 同时导致请求出现较大的波动，造成不好的体验，所以在结构中有两个map, 一个为只读map, 另一个为可写map, 读取缓存时只读取只读map，更新缓存时, 将所有缓存写入到可写map, 更新写入完成后, 将最新缓存map加速赋值给只读map, 同时清空更新map, 伪代码如下:

package main

import "sync"

type localCache struct {
   lock   sync.RWMutex
   Cache  map[string]interface{}
   update map[string]interface{}
}

func (c *localCache) Get(key string) interface{} {
   c.lock.RLock()
   defer c.lock.RUnlock()
   
   cache, ok := c.Cache[key]
   if ok {
      return cache
   }
   return nil
}

func (c *localCache) Set(key string, val interface{}) {
   c.update[key] = val
}

// 初始化待更新数据的长度
func (c *localCache) MakeUpdateCache(len int) {
   c.update = make(map[string]interface{}, len)
}

func (c *localCache) Rehash() {
   c.lock.RLock()
   defer c.lock.RUnlock()

   c.Cache = c.update
   c.MakeUpdateCache(0)
}

这样只会在更新db时不会频繁的对于一个map进行扩容操作, 同时可以保证Cache字段中不会存在历史的脏数据。只有在执行Rehash()方法完成时才会执行内存分配的动作, 而且在读多写少的情况下, RWMutex性能可以达到互斥锁的8倍左右(参考Go 语言高性能编程)。这个是基于渐进式rehash可以带给我的关于自己做本地缓存的思考, 但是这个方案我自己并未实践, 所以各种细节考虑还有待改进, 也欢迎讨论。

动态内存分配

分享一个工作中的问题, 有一个case现象是当请求并发量比较大的时候会出现cpu使用率飙升, 查看了相关接口的业务逻辑相对比较简单, 从磁盘中加载数据到内存map中, 然后返回map中的key。然后开始查看系统调用关系, 发现有非常对的cpu耗时在php的内存分配上mmap, debug发现使用了php数组的+=操作进行对象合并, 而且每次操作数据比较大, 所以造成了cpu使用率较高, 解决方案也很简单, 可以把类似的操作改为共享内存操作, 通过lstat调用我们已经获取需要保存文件的总体大小, 我们只需要一次性开辟一块内存存储即可, 可以使用共享内存操作:

$shmid = shmop_open($systemid, $mode, $permissions, $size);

参数四是需要开辟的内存大小, 即可减少内存分配的次数, 降低频繁的内存分配。

数据结构

Redis中实现了很多优秀的数据结构, 有些不仅仅是典型的实现, 而是根据实际需求做了改进, 如Skiplist中加入了向后的指针等。关于数据结构及算法设计可能在大家平常的工作中并不会有太多时间去深入思考与应用, 业界也有比较成熟, 可靠的方案可借鉴, 但其实如果抛开现有的方案, 我们需要的数据结构及算法在工作中处处可见, 如父子节点的组织构造与查找, 自己构造权重队列等场景。

前一段时间工作有一个场景是给客户端下发一组原因选项, 每个选项下面可能存在一个二级选项, 所以这个场景可以是一个无限极分类, 初次的实现中是通过递归调用来查找的。后面优化使用树的深度遍历来查找, 个人思考是能够使用合适的数据结构实现不同的业务有助于提升自己的思维能力。