redis源码解读-大体的执行流程&一些常用的数据结构以及数据类型

Posted 2021-08-15 _微风轻起

看了一些关于redis 的相关文章，例如redis为什么这么快、redis的原理、redis的数据结构这些，但其只是从整体的结构来说明，并没有梳理源码的具体流程，但我不是很喜欢一些黑盒的东西，所以我们这一篇就通过跑redis的源码，来追踪redis源码中的一些数据结构。这篇文章的源码是基于redis 3.0版本，同时源码是直接从github上面clone下来的别人已经处理好的redis代码(windows平台)，地址为：https://github.com/htw0056/redis-3.0-annotated-cmake-in-clion。感谢这位前辈。

​ 首先我们启动服务端，然后再启动客户端，通过客户端输入命令我们来跟踪其的主要结构，下面就正式开始。

​ 我们通过客户端来设置值，然后来追踪其的执行。

一、redisClient关联内容

​ 其首先是从redis.c的main还是开始的，然后通过在main函数中调用aeMain(aeEventLoop *eventLoop)来处理事件(redis使用的是IO多路复用epoll)。当然这里以及之后会进行各种内存、数据结构的初始化等。我们就不具体分析这种了（对于c语言我也只是能看懂大概的内容，具体细节也不是很明白）。但我们通过源码debug还是能明白大体的结构的。

1、整体执行流程

/*
 * 事件处理器的主循环
 */
void aeMain(aeEventLoop *eventLoop) {

    eventLoop->stop = 0;

    while (!eventLoop->stop) {

        // 如果有需要在事件处理前执行的函数，那么运行它
        if (eventLoop->beforesleep != NULL)
            eventLoop->beforesleep(eventLoop);

        // 开始处理事件
        aeProcessEvents(eventLoop, AE_ALL_EVENTS);
    }
}

​ 下面我们就直接到processCommand方法，来看其对于输入命令的具体执行。

int processCommand(redisClient *c) {
		.............
  		
    // 如果设置了最大内存，那么检查内存是否超过限制，并做相应的操作
    if (server.maxmemory) {
        // 如果内存已超过限制，那么尝试通过删除过期键来释放内存
        int retval = freeMemoryIfNeeded();
        // 如果即将要执行的命令可能占用大量内存（REDIS_CMD_DENYOOM）
        // 并且前面的内存释放失败的话
        // 那么向客户端返回内存错误
        if ((c->cmd->flags & REDIS_CMD_DENYOOM) && retval == REDIS_ERR) {
            flagTransaction(c);
            addReply(c, shared.oomerr);
            return REDIS_OK;
        }
    }
		........
    /* Don't accept write commands if this is a read only slave. But
     * accept write commands if this is our master. */
    // 如果这个服务器是一个只读 slave 的话，那么拒绝执行写命令
    if (server.masterhost && server.repl_slave_ro &&
        !(c->flags & REDIS_MASTER) &&
        c->cmd->flags & REDIS_CMD_WRITE)
    {
        addReply(c, shared.roslaveerr);
        return REDIS_OK;
    }
		............
        // 执行命令
        call(c,REDIS_CALL_FULL);
		..........
    return REDIS_OK;
}

​ 这里我们省略了很多内容，值保留了部分内容。

2、结构体内容

1）、redisClient结构体

​ 首先我们来看下redisClient结构体，这个就是redis的客户端链接命令处理：

typedef struct redisClient {

    // 套接字描述符
    int fd;

    // 当前正在使用的数据库
    redisDb *db;

    // 当前正在使用的数据库的 id （号码）
    int dictid;

    // 客户端的名字
    robj *name;             /* As set by CLIENT SETNAME */
		.........

} redisClient;

​ 这里是直接有redisDB，然后dictid是表示当前使用的是哪个redisDB，例如当前就是使用的0号库。然后在redisDb中，就有真正存放添加的数据了。

2）、redisDb结构体

typedef struct redisDb {

    // 数据库键空间，保存着数据库中的所有键值对
    dict *dict;                 /* The keyspace for this DB */

    // 键的过期时间，字典的键为键，字典的值为过期事件 UNIX 时间戳
    dict *expires;              /* Timeout of keys with a timeout set */
		......
    // 数据库号码
    int id;                     /* Database ID */

    // 数据库的键的平均 TTL ，统计信息
    long long avg_ttl;          /* Average TTL, just for stats */

} redisDb;

​ 这个就是redis数据库的结构体了，类似于mysql的数据库的概念，不过其的数据库名称就是id标识，表示是第几号库。

3）、dict结构体

typedef struct dict {

    // 类型特定函数
    dictType *type;

    // 私有数据
    void *privdata;

    // 哈希表
    dictht ht[2];

    // rehash 索引
    // 当 rehash 不在进行时，值为 -1
    int rehashidx; /* rehashing not in progress if rehashidx == -1 */

    // 目前正在运行的安全迭代器的数量
    int iterators; /* number of iterators currently running */

} dict;

​ 这个就是字典表。然后数据就是存在两张hash表中ht[2]中**（redis不管要存入的是哪种类型例如string、list、hash这些，都是放在这个hash表中**、（然后这里就是文章上说的，redis为什么这么快的第二个答案了(因为使用hash结构，其的查找会很快，第一个快的原因我认为是使用的内存），同时要注意这些类型存在redis内部又是其他具体设置的结构体、例如LINKEDLIST常规链表、ZIPLIST压缩列表、SKIPLIST跳表），这个我们后面再来看。

​ 然后这里dictht ht[2]之所以是两张表，是用来扩容使用的，在渐进式hash扩容期间其是会使用两张表的，然后在ht[0]中获取不到，就会去ht[1]中找**(关于渐进式hash我们后面也会说明)、(这个也可以是速度快的第三个原因)**。然后这里的rehashidx就用来标识当前是不是在渐进式hash扩容期间。

4）、dictht &dictEntry 结构体

typedef struct dictht {
    
    // 哈希表数组
    dictEntry **table;

    // 哈希表大小
    unsigned long size;
    
    // 哈希表大小掩码，用于计算索引值
    // 总是等于 size - 1
    unsigned long sizemask;

    // 该哈希表已有节点的数量
    unsigned long used;

} dictht;

typedef struct dictEntry {
    
    // 键
    void *key;

    // 值
    union {
        void *val;
        uint64_t u64;
        int64_t s64;
    } v;

    // 指向下个哈希表节点，形成链表
    struct dictEntry *next;

} dictEntry;

​ 这个就是hash表数组(客户端的设置获取命令最终都要落到table来。)，然后used表示当前包含的节点，而dictEntry就是对应的实体名称。

​ 可以看到目前我们的key是有30个。

5）、整体数据介绍

​ 然后我们再来看下redisClient关联的db数据：

​ 可以看到其当前使用的个数是30个，同时其也要扩容了。但现在还没有扩容rehashidx是-1。

3、processCommand执行步骤

1）、查询检查命令

​ 这里就是检查查看有没有我们当前输入的命令，如果没有就给出提示返回unknown command。

c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
if (!c->cmd) {
    // 没找到指定的命令
    flagTransaction(c);
    addReplyErrorFormat(c,"unknown command '%s'",
        (char*)c->argv[0]->ptr);
    return REDIS_OK;
}

struct redisCommand *lookupCommand(sds name) {
    return dictFetchValue(server.commands, name);
}

void *dictFetchValue(dict *d, const void *key) {
    dictEntry *he;

    // T = O(1)
    he = dictFind(d,key);

    return he ? dictGetVal(he) : NULL;
}

​ 我们目前是set命令。

2）、server.commands填充逻辑

​ 首这个字典的参数化就是在populateCommandTable方法遍历添加的。

void populateCommandTable(void) {
    int j;

    // 命令的数量
    int numcommands = sizeof(redisCommandTable)/sizeof(struct redisCommand);

    for (j = 0; j < numcommands; j++) {
        
        // 指定命令
        struct redisCommand *c = redisCommandTable+j;

        // 取出字符串 FLAG
        char *f = c->sflags;
			...........
        // 将命令关联到命令表
        retval1 = dictAdd(server.commands, sdsnew(c->name), c);
		.........
    }
}

struct redisCommand redisCommandTable[] = {
    {"get",getCommand,2,"r",0,NULL,1,1,1,0,0},
    {"set",setCommand,-3,"wm",0,NULL,1,1,1,0,0},
    {"setnx",setnxCommand,3,"wm",0,NULL,1,1,1,0,0},
    {"setex",setexCommand,4,"wm",0,NULL,1,1,1,0,0},
    {"psetex",psetexCommand,4,"wm",0,NULL,1,1,1,0,0},
    {"append",appendCommand,3,"wm",0,NULL,1,1,1,0,0},
    {"strlen",strlenCommand,2,"r",0,NULL,1,1,1,0,0},
    {"del",delCommand,-2,"w",0,NULL,1,-1,1,0,0},
    {"exists",existsCommand,2,"r",0,NULL,1,1,1,0,0},
    {"setbit",setbitCommand,4,"wm",0,NULL,1,1,1,0,0},
    {"getbit",getbitCommand,3,"r",0,NULL,1,1,1,0,0},
    {"setrange",setrangeCommand,4,"wm",0,NULL,1,1,1,0,0},
    {"getrange",getrangeCommand,4,"r",0,NULL,1,1,1,0,0},
    {"substr",getrangeCommand,4,"r",0,NULL,1,1,1,0,0},
    {"incr",incrCommand,2,"wm",0,NULL,1,1,1,0,0},
    {"decr",decrCommand,2,"wm",0,NULL,1,1,1,0,0},
    {"mget",mgetCommand,-2,"r",0,NULL,1,-1,1,0,0},
    {"rpush",rpushCommand,-3,"wm",0,NULL,1,1,1,0,0},
    {"lpush",lpushCommand,-3,"wm",0,NULL,1,1,1,0,0},
    {"rpushx",rpushxCommand,3,"wm",0,NULL,1,1,1,0,0},
    {"lpushx",lpushxCommand,3,"wm",0,NULL,1,1,1,0,0},
  		............
    {"pfdebug",pfdebugCommand,-3,"w",0,NULL,0,0,0,0,0}
};

​ 这里面就是所有的redis命令。

3）、字典添加的逻辑(dictAddRaw)

int dictAdd(dict *d, void *key, void *val)
{
    // 尝试添加键到字典，并返回包含了这个键的新哈希节点
    // T = O(N)
    dictEntry *entry = dictAddRaw(d,key);

    // 键已存在，添加失败
    if (!entry) return DICT_ERR;

    // 键不存在，设置节点的值
    // T = O(1)
    dictSetVal(d, entry, val);

    // 添加成功
    return DICT_OK;
}

dictEntry *dictAddRaw(dict *d, void *key)
{
    int index;
    dictEntry *entry;
    dictht *ht;
    // 如果条件允许的话，进行单步 rehash
    // T = O(1)
    if (dictIsRehashing(d)) _dictRehashStep(d);

    /* Get the index of the new element, or -1 if
     * the element already exists. */
    // 计算键在哈希表中的索引值
    // 如果值为 -1 ，那么表示键已经存在
    // T = O(N)
    if ((index = _dictKeyIndex(d, key)) == -1)
        return NULL;

    // T = O(1)
    /* Allocate the memory and store the new entry */
    // 如果字典正在 rehash ，那么将新键添加到 1 号哈希表
    // 否则，将新键添加到 0 号哈希表
    ht = dictIsRehashing(d) ? &d->ht[1] : &d->ht[0];
    // 为新节点分配空间
    entry = zmalloc(sizeof(*entry));
    // 将新节点插入到链表表头
    entry->next = ht->table[index];
    ht->table[index] = entry;
    // 更新哈希表已使用节点数量
    ht->used++;

    /* Set the hash entry fields. */
    // 设置新节点的键
    // T = O(1)
    dictSetKey(d, entry, key);

    return entry;
}

​ 这里具体有3步：首先看需不需要单步渐进hash赋值，如果需要(扩容的时候才需要)，就渐进hash(其的步长是1(也就是一次只一定hash表的一个位置的节点链表到另一张表))，然后创建分配dictEntry，再计算其的hash表index使用头插法将其添加hash表中。这个是命令的字典表。当我们的设置字典表也是类似这个逻辑。

4）、内存清除

/* Handle the maxmemory directive.
 *
 * First we try to free some memory if possible (if there are volatile
 * keys in the dataset). If there are not the only thing we can do
 * is returning an error. */
// 如果设置了最大内存，那么检查内存是否超过限制，并做相应的操作
if (server.maxmemory) {
    // 如果内存已超过限制，那么尝试通过删除过期键来释放内存
    int retval = freeMemoryIfNeeded();
    // 如果即将要执行的命令可能占用大量内存（REDIS_CMD_DENYOOM）
    // 并且前面的内存释放失败的话
    // 那么向客户端返回内存错误
    if ((c->cmd->flags & REDIS_CMD_DENYOOM) && retval == REDIS_ERR) {
        flagTransaction(c);
        addReply(c, shared.oomerr);
        return REDIS_OK;
    }
}

​ 在正式执行命令设置内容之前，我们需要检查是不是已经达到最大内存了，如果到了，我们就需要根据内存淘汰策略其起来一些值了：

int freeMemoryIfNeeded(void) {
    size_t mem_used, mem_tofree, mem_freed;
    int slaves = listLength(server.slaves);
		............
      // 如果占用内存比 maxmemory 要大，但是 maxmemory 策略为不淘汰，那么直接返回
    if (server.maxmemory_policy == REDIS_MAXMEMORY_NO_EVICTION)
        return REDIS_ERR; /* We need to free memory, but policy forbids. */
    /* Compute how much memory we need to free. */
    // 计算需要释放多少字节的内存
    mem_tofree = mem_used - server.maxmemory;
    // 初始化已释放内存的字节数为 0
    mem_freed = 0;
    // 根据 maxmemory 策略，
    // 遍历字典，释放内存并记录被释放内存的字节数
    while (mem_freed < mem_tofree) {
        int j, k, keys_freed = 0;
        // 遍历所有字典
        for (j = 0; j < server.dbnum; j++) {
            long bestval = 0; /* just to prevent warning */
            sds bestkey = NULL;
            dictEntry *de;
            redisDb *db = server.db+j;
            dict *dict;

            if (server.maxmemory_policy == REDIS_MAXMEMORY_ALLKEYS_LRU ||
                server.maxmemory_policy == REDIS_MAXMEMORY_ALLKEYS_RANDOM)
            {
                // 如果策略是 allkeys-lru 或者 allkeys-random 
                // 那么淘汰的目标为所有数据库键
                dict = server.db[j].dict;
            } else {
                // 如果策略是 volatile-lru 、 volatile-random 或者 volatile-ttl 
                // 那么淘汰的目标为带过期时间的数据库键
                dict = server.db[j].expires;
            }

            // 跳过空字典
            if (dictSize(dict) == 0) continue;

            /* volatile-random and allkeys-random policy */
            // 如果使用的是随机策略，那么从目标字典中随机选出键
            if (serverTiDB内核—源码剖析解读

 曹工说Redis源码-- redis server 主循环大体流程解析
 Spark源码解读--spark.textFile()读取流程
 前端工程化9：Webpack构建流程分析，Webpack5源码解读
 PolarDB-X 源码解读系列：DML 之 INSERT IGNORE 流程
 ElasticSearchEs 源码之 SearchService 源码解读