公众号「码哥字节」，拥抱硬核技术和对象，面向人民币编程。

整体架构

当你熟悉我的整体架构和每个模块，遇到问题才能直击本源，直捣黄龙，一笑破苍穹。我的核心模块如图。

• Client 客户端，官方提供了 C 语言开发的客户端，可以发送命令，性能分析和测试等。
• 网络层事件驱动模型，基于 I/O 多路复用，封装了一个短小精悍的高性能 ae 库，全称是 a simple event-driven programming library。
• 在 ae 这个库里面，我通过 aeApiState 结构体对 epoll、select、kqueue、evport四种 I/O 多路复用的实现进行适配，让上层调用方感知不到在不同操作系统实现 I/O 多路复用的差异。
• Redis 中的事件可以分两大类：一类是网络连接、读、写事件；另一类是时间事件，比如定时执行 rehash 、RDB 内存快照生成，过期键值对清理操作。
• 命令解析和执行层，负责执行客户端的各种命令，比如 SET、DEL、GET等。
• 内存分配和回收，为数据分配内存，提供不同的数据结构保存数据。
• 持久化层，提供了 RDB 内存快照文件 和 AOF 两种持久化策略，实现数据可靠性。
• 高可用模块，提供了副本、哨兵、集群实现高可用。
• 监控与统计，提供了一些监控工具和性能分析工具，比如监控内存使用、基准测试、内存碎片、bigkey 统计、慢指令查询等。

掌握了整体架构和模块后，接下来进入 src 源码目录，使用如下指令执行 redis-server可执行程序启动 Redis。

./redis-server ../redis.conf

每个被启动的服务我都会抽象成一个 redisServer，源码定在server.h 的redisServer 结构体。

这个结构体包含了存储键值对的数据库实例、redis.conf 文件路径、命令列表、加载的 Modules、网络监听、客户端列表、RDB AOF 加载信息、配置信息、RDB 持久化、主从复制、客户端缓存、数据结构压缩、pub/sub、Cluster、哨兵等一些列 Redis 实例运行的必要信息。

结构体字段很多，不再一一列举，部分核心字段如下。

truct redisServer {
    pid_t pid;  /* 主进程 pid. */
    pthread_t main_thread_id; /* 主线程 id */
    char *configfile;  /*redis.conf 文件绝对路径*/
    redisDb *db; /* 存储键值对数据的 redisDb 实例 */
   int dbnum;  /* DB 个数 */
    dict *commands; /* 当前实例能处理的命令表，key 是命令名，value 是执行命令的入口 */
    aeEventLoop *el;/* 事件循环处理 */
    int sentinel_mode;  /* true 则表示作为哨兵实例启动 */

   /* 网络相关 */
    int port;/* TCP 监听端口 */
    list *clients; /* 连接当前实例的客户端列表 */
    list *clients_to_close; /* 待关闭的客户端列表 */

    client *current_client; /* 当前执行命令的客户端*/
};

数据存储原理

其中redisDb *db指针非常重要，它指向了一个长度为 dbnum（默认 16）的 redisDb 数组，它是整个存储的核心，我就是用这玩意来存储键值对。

redisDb

redisDb 结构体定义如下。

typedef struct redisDb {
    dict *dict;
    dict *expires;
    dict *blocking_keys;
    dict *ready_keys;
    dict *watched_keys;
    int id;
    long long avg_ttl;
    unsigned long expires_cursor;
    list *defrag_later;
    clusterSlotToKeyMapping *slots_to_keys;
} redisDb;

dict 和 expires

• dict 和 expires 是最重要的两个属性，底层数据结构是字典，分别用于存储键值对数据和 key 的过期时间。
• expires，底层数据结构是 dict 字典，存储每个 key 的过期时间。

?

MySQL：“为什么分开存储？”

好问题，之所以分开存储，是因为过期时间并不是每个 key 都会设置，它不是键值对的固有属性，分开后虽然需要两次查找，但是能节省内存开销。

blocking_keys 和 ready_keys

底层数据结构是 dict 字典，主要是用于实现 BLPOP 等阻塞命令。

当客户端使用 BLPOP 命令阻塞等待取出列表元素的时候，我会把 key 写到 blocking_keys 中，value 是被阻塞的客户端。

当下一次收到 PUSH 命令执时，我会先检查 blocking_keys 中是否存在阻塞等待的 key，如果存在就把 key 放到 ready_keys 中，在下一次 Redis 事件处理过程中，会遍历 ready_keys 数据，并从 blocking_keys 中取出阻塞的客户端响应。

slots_to_keys

仅用于 Cluster 模式，当使用 Cluster 模式的时候，只能有一个数据库 db 0。slots_to_keys 用于记录 cluster 模式下，存储 key 与哈希槽映射关系的数组。

dict

Redis 使用 dict 结构来保存所有的键值对（key-value）数据，这是一个散列表，所以 key 查询时间复杂度是 O(1) 。

所谓散列表，我们可以类比 Java 中的 HashMap，其实就是一个数组，数组的每个元素叫做哈希桶。

dict 结构体源码在 dict.h 中定义。

struct dict {
    dictType *type;

    dictEntry **ht_table[2];
    unsigned long ht_used[2];

    long rehashidx;

    int16_t pauserehash;
    signed char ht_size_exp[2];
};

dict 的结构体里，有 dictType *type，**ht_table[2]，long rehashidx 三个很重要的结构。

• type 存储了 hash 函数，key 和 value 的复制等函数；
• ht_table[2]，长度为 2 的数组，默认使用 ht_table[0] 存储键值对数据。我会使用 ht_table[1] 来配合实现渐进式 reahsh 操作。
• rehashidx 是一个整数值，用于标记是否正在执行 rehash 操作，-1 表示没有进行 rehash。如果正在执行 rehash，那么其值表示当前 rehash 操作执行的 ht_table[1] 中的 dictEntry 数组的索引。
• pauserehash 表示 rehash 的状态，大于 0 时表示 rehash 暂停了，小于 0 表示出错了。
• ht_used[2]，长度为 2 的数组，表示每个哈希桶存储了多少个 键值对实体（dictEntry），值越大，哈希冲突的概率越高。
• ht_size_exp[2]，每个散列表的大小，也就是哈希桶个数。

重点关注 ht_table 数组，数组每个位置叫做哈希桶，就是这玩意保存了所有键值对，每个哈希桶的类型是 dictEntry。

?

MySQL：“Redis 支持那么多的数据类型，哈希桶咋保存？”

他的玄机就在 dictEntry 中，每个 dict 有两个 ht_table，用于存储键值对数据和实现渐进式 rehash。

dictEntry 结构如下。

typedef struct dictEntry {
    void *key;
    union {
       // 指向实际 value 的指针
        void *val;
        uint64_t u64;
        int64_t s64;
        double d;
    } v;
    // 散列表冲突生成的链表
    struct dictEntry *next;
    void *metadata[];
} dictEntry;

• *key 指向键值对的键的指针，指向一个 sds 对象，key 都是 string 类型。
• v 是键值对的 value 值，是个 union（联合体），当它的值是 uint64_t、int64_t 或 double 数字类型时，就不再需要额外的存储，这有利于减少内存碎片。（为了节省内存操碎了心）当值为非数字类型，就是用 val 指针存储。
• *next指向另一个 dictEntry 结构， 多个 dictEntry 可以通过 next 指针串连成链表， 从这里可以看出， ht_table 使用链地址法来处理键碰撞： 当多个不同的键拥有相同的哈希值时，哈希表用一个链表将这些键连接起来。

哈希桶并没有保存值本身，而是指向具体值的指针，从而实现了哈希桶能存不同数据类型的需求。

redisObject

dictEntry 的 *val 指针指向的值实际上是一个 redisObject 结构体，这是一个非常重要的结构体。

我的 key 是字符串类型，而 value 可以是 String、Lists、Set、Sorted Set、Hashes 等数据类型。

键值对的值都被包装成 redisObject 对象， redisObject 在 server.h 中定义。

typedef struct redisObject {
    unsigned type:4;
    unsigned encoding:4;
    unsigned lru:LRU_BITS;
    int refcount;
    void *ptr;
} robj;

• type： 记录了对象的类型，string、set、hash 、Lis、Sorted Set 等，根据该类型来确定是哪种数据类型，这样我才知道该使用什么指令执行嘛。
• encoding：编码方式，表示 ptr 指向的数据类型具体数据结构，即这个对象使用了什么数据结构作为底层实现保存数据。同一个对象使用不同编码内存占用存在明显差异，节省内存，这玩意功不可没。
• lru:LRU_BITS：LRU 策略下对象最后一次被访问的时间，如果是 LFU 策略，那么低 8 位表示访问频率，高 16 位表示访问时间。
• refcount ：表示引用计数，由于 C 语言并不具备内存回收功能，所以 Redis 在自己的对象系统中添加了这个属性，当一个对象的引用计数为 0 时，则表示该对象已经不被任何对象引用，则可以进行垃圾回收了。
• ptr 指针：指向值的指针，对象的底层实现数据结构。

如图 1-11 是由 redisDb、dict、dictEntry、redisObejct 关系图：

注意，一开始的时候，我只使用 ht_table[0] 这个散列表读写数据，ht_table[1] 指向 NULL，当这个散列表容量不足，触发扩容操作，这时候就会创建一个更大的散列表 ht_table[1]。

接着我会使用渐进式 rehash 的方式和定期迁移的方式将 ht_table[0] 的数据迁移到 ht_table[1] 上，全部迁移完成后，再修改下指针，让 ht_table[0] 指向扩容后的散列表，回收掉原来的散列表，ht_table[1] 再次指向 NULL。

一条指令的执行过程

?

MySQL：“知道了整体架构和各个模块后，一条 Redis 命令是如何执行呢？”

想要了解一个技术的本质，先从整体再到细节，切不可不见森林就去看叶子，所以我先梳理出一个关键执行流程给你，防止陷入细节不能自拔。

跟着流程图，我再详细介绍每一步执行详细步骤，本流程图重点展示的是在开启 I/O 多线程模型的情况下的指令执行过程。

先看 server.c 中的 main() 函数，里面有几个很重要的几个步骤。

• initServerConfig()：初始化 Redis 服务的各种配置。
• initServer()：初始化服务，分配运行需要的数据结构，设置监听 socket 等。
• aeMain()：监听新连接的事件循环，通过事件驱动模块 ae 接收客户端发起的请求。

创建连接

Redis 初始化完成，默认会在 6379 端口监可读事件，通过事件驱动模块 ae 接收客户端发起的请求，这是一个基于 I/O 多路复用的 while 无限循环。

源代码在 server.c的 createSocketAcceptHandler函数中。

• 初始化的时候，createSocketAcceptHandler()函数只监听了可读事件，并将 acceptTcpHandler()函数作为处理该事件的回调函数。
• acceptTcpHandler() 函数的核心就是一个 while 循环，里面是处理建立连接的请求。
• 执行完 acceptTcpHandler() 函数后，一条 Redis 客户端连接就创建完成了。

延迟读取

客户端发来SET key 码哥字节命令请求，触发可读事件，实际上就是触发了 connSocketEventHandler() 函数，该函数的内部会处理可读事件和可写事件，默认先处理可读事件，再处理可写事件（正常情况下是先有请求可读事件发生，处理好读事件之后，才能响应客户端，这个时候再处理可写事件）。

先看可读事件的处理，实际就是进入 readQueryFromClient()函数执行流程，这个流程会判断是否启用 I/O 多线程来选择不同分支处理。

开启 I/O 多线程模式

主线程首次调用 readQueryFromClient() 函数时会先执行 postponeClientRead()将可读事件的 client 放入 redisServer.clients_pending_read 列表。

在主线程每次进入 aeApiPoll()函数阻塞等待可读可写事件之前，会先调用 beforeSleep()函数，这个函数会调用 handleClientsWithPendingReadsUsingThreads()函数，他的核心逻辑是把主函数上次调用 readQueryFromClient() 函数把 client 写到redisServer.clients_pending_read 队列的 client 按照 Round Robbin 的方式分配到每个 I/O 线程关联的 io_threads_list队列，I/O 线程消费该队列进行请求读取和命令解析。

handleClientsWithPendingReadsUsingThreads 源代码如下，我加了关键注释。

int handleClientsWithPendingReadsUsingThreads(void) {
    // 1. I/O 多线程模型是否启用，未启用的话停止后续执行。
    if (!server.io_threads_active || !server.io_threads_do_reads) return 0;
    int processed = listLength(server.clients_pending_read);
    if (processed == 0) return 0;

    /* 2. 将 server.clients_pending_read 中的 client 分配到不同的 io_threads_list */
    listIter li;
    listNode *ln;
    listRewind(server.clients_pending_read,&li);
    int item_id = 0;
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        int target_id = item_id % server.io_threads_num;
        listAddNodeTail(io_threads_list[target_id],c);
        item_id++;
    }

    /* 3. 设置全局变量 io_threads_op 为 IO_THREADS_OP_READ 告诉 I/O 线程此次处理的是
     * 可读事件。
     */
    io_threads_op = IO_THREADS_OP_READ;
    for (int j = 1; j < server.io_threads_num; j++) {
        int count = listLength(io_threads_list[j]);
        setIOPendingCount(j, count);
    }

    /* 4. 主线程处理 io_threads_list[0] 中的任务，调用 readQueryFromClient 读取数据并解析命令，完成清空 io_threads_list[0] 列表 */
    listRewind(io_threads_list[0],&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        readQueryFromClient(c->conn);
    }
    listEmpty(io_threads_list[0]);

    /* 5. 主线程阻塞等待其他 I/O 线程完成数据读取和命令解析。 */
    while(1) {
        unsigned long pending = 0;
        for (int j = 1; j < server.io_threads_num; j++)
            pending += getIOPendingCount(j);
        if (pending == 0) break;
    }

    // 6. 全局变量 io_threads_op 设置成 IO_THREADS_OP_IDLE 表示当前 I/O 线程空闲。
    io_threads_op = IO_THREADS_OP_IDLE;

    /* 7. 主线程从 clients_pending_read 取出 client 并执行已经读取完成并解析好的命令*/
    while(listLength(server.clients_pending_read)) {
        // 省略部分代码

        if (processPendingCommandAndInputBuffer(c) == C_ERR) {
            continue;
        }
        // 省略部分代码
    }

    server.stat_io_reads_processed += processed;

    return processed;
}

未开启 I/O 多线程模型

主线程自己完成读取命令、解析命令、执行命令、发送结果给客户端全部流程。

命令读取和解析

主线程首次调用 readQueryFromClient() 会通过 postponeClientRead()把可读事件延迟到 I/O 线程处理，就是把可读事件的 client 放到 server.clients_pending_read 中。再按照 Round Robbin 的方式把 client 分配到每个 I/O 线程关联的 io_threads_list队列

主线程和 I/O 线程从绑定的 io_treads_list任务队列中取出任务并处理。主线程和 I/O 线程再次进入 readQueryFromClient()流程。

需要注意的是，这次执行readQueryFromClient()前， client 状态已经被设置成 CLIENT_PENDING_READ，不会再次加入 clients_pending_read队列，而是进入真正的执行流程，读取 Socket 并解析命令。

主线程在完成 io_threads_list[0] client 的读取和解析之后，会阻塞等待全部 I/O 线程完成等到全部命令解析完成，才会真正的执行命令。

命令执行

主线程等所有 I/O 线程完成 Socket 读取和命令解析，接下来到了最重要的一点，执行命令。

回到 handleClientsWithPendingReadsUsingThreads函数，在完成了任务分配和命令读取和解析之后，主线程会进入一个 while 循环，从 server.clients_pending_read 队列中，每取出一个 client 就调用一次次 processPendingCommandAndInputBuffer()函数执行这个 client 已经解析好的命令。

/* 7. 主线程从 clients_pending_read 取出 client 并执行已经读取完成并解析好的命令*/
while(listLength(server.clients_pending_read)) {
    // 省略部分代码

    if (processPendingCommandAndInputBuffer(c) == C_ERR) {
        continue;
    }
    // 省略部分代码
}

函数内部会调用 processCommandAndResetClient()函数执行实际的命令。

int processPendingCommandAndInputBuffer(client *c) {
    if (c->flags & CLIENT_PENDING_COMMAND) {
        c->flags &= ~CLIENT_PENDING_COMMAND;
        // 主线程执行命令
        if (processCommandAndResetClient(c) == C_ERR) {
            return C_ERR;
        }
    }

    // 省略部分源码
    return C_OK;
}

processCommandAndResetClient 函数源码如下。

int processCommandAndResetClient(client *c) {
    // 省略部分源码
    if (processCommand(c) == C_OK) {
        commandProcessed(c);
        updateClientMemUsageAndBucket(c);
    }
 // 省略部分源码
    return deadclient ? C_ERR : C_OK;
}

重点看 processCommand(c)函数，他的核心逻辑是从调用lookupCommand()函数从 server.commands这个字典中查找命令对应的 redisCommand实例。

经过系列检查，不通过就执行 rejectCommandFormat() 函数给客户端返回错误信息。

通过的话就调用 c->cmd->proc()函数，处理真正的命令，比如 SET 命令， proc() 就指向 setCommand() 函数。

 /* Exec the command */
if (c->flags & CLIENT_MULTI &&
    c->cmd->proc != execCommand &&
    c->cmd->proc != discardCommand &&
    c->cmd->proc != multiCommand &&
    c->cmd->proc != watchCommand &&
    c->cmd->proc != quitCommand &&
    c->cmd->proc != resetCommand)
{
    // 1. 命令入队
    queueMultiCommand(c, cmd_flags);
    // 2. 入队后响应客户端 "+QUEUED" 字符串
    addReply(c,shared.queued);
} else {
    // 3. 不需要入队，直接执行的命令调用 call() 函数
    call(c,CMD_CALL_FULL);
    c->woff = server.master_repl_offset;
    if (listLength(server.ready_keys))
        handleClientsBlockedOnKeys();
}

函数执行流程图如下。

回头再看 processCommandAndResetClient 函数，发现 return deadclient ? C_ERR : C_OK; 根据成功返回 1，错误返回 0。

?

MySQL ：“并没有看到将命令产生的返回值写回客户端的代码，你是如何将命令产生的返回值写回客户端的？”

通过流程图你知道，在开启 I/O 多线程模型的时候，我是通过 I/O 线程将命令的返回值写回客户端的。

虽然调用 redisCommand->proc()的时候没有返回值，但是我在 proc() 函数里将返回值写到一个专门存放返回值的地方让 I/O 线程去取值并返回客户端。

响应结果给客户端

我用 String 类型的 GET命令为例，源码文件是 t_string.c。getCommand 啥都不干，直接调用 getGenericCommand函数。

int getGenericCommand(client *c) {
    robj *o;
 // 1. 从数据库中查询 key 对应的 value 值，并赋值给 o.
    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.null[c->resp])) == NULL)
        return C_OK;
 // 一些类型校验
    if (checkType(c,o,OBJ_STRING)) {
        return C_ERR;
    }
 // 对返回值进行编码并返回
    addReplyBulk(c,o);
    return C_OK;
}

void getCommand(client *c) {
    getGenericCommand(c);
}

重点就在于 addReplyBulk(c,o)，执行命令完毕后，进入响应客户端阶段，主线程调用 addReply函数把执行结果响应给客户端。

void addReply(client *c, robj *obj) {
    if (prepareClientToWrite(c) != C_OK) return;
    ...
}

内部调用 prepareClientToWrite把执行结果放到 clients_pending_write 可写队列中。

在进入下一次事件循环时， beforeSleep() 函数内部会调用 handleClientsWithPendingWritesUsingThreads 函数把 clients_pending_write队列任务分配给 I/O 线程和主线程。

完成任务分配之后，I/O 线程和主线程会调用 writeToClient函数把命令的执行结果发送到客户端，writeToClient() 函数的核心是一个 while 循环，内部会不断调用 _writeToClient()函数，往底层的 Scoket 连接里面写数据。

详细步骤我在源代码中加了注释。

int handleClientsWithPendingWritesUsingThreads(void) {
    // 1. 没有客户端需要处理，return 0。
    int processed = listLength(server.clients_pending_write);
    if (processed == 0) return 0;

    // 2. 如果没有启用 I/O 线程或者只有少量 client 需要处理，主线程同步处理，不使用 I/O 线程。
    if (server.io_threads_num == 1 || stopThreadedIOIfNeeded()) {
        return handleClientsWithPendingWrites();
    }

    /* 3. 开启了I/O 多线程模式，但是没有激活的话则调用 startThreadedIO 激活 */
    if (!server.io_threads_active) startThreadedIO();

    /* 4. 将 clients_pending_write 队列 client 分配到不同io_threads_list 列表中，主线程负责 io_threads_list[0] 的任务
    */
    listIter li;
    listNode *ln;
    listRewind(server.clients_pending_write,&li);
    int item_id = 0;
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        c->flags &= ~CLIENT_PENDING_WRITE;
  // 省略部分代码

        int target_id = item_id % server.io_threads_num;
        listAddNodeTail(io_threads_list[target_id],c);
        item_id++;
    }

    /* 5. 修改全局变量，标识可写 */
    io_threads_op = IO_THREADS_OP_WRITE;
    for (int j = 1; j < server.io_threads_num; j++) {
        int count = listLength(io_threads_list[j]);
        setIOPendingCount(j, count);
    }

    /* 6. 主线程调用 writeToClient 把 io_threads_list[0] 的 client 执行结果写回给客户端  */
    listRewind(io_threads_list[0],&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);
        writeToClient(c,0);
    }
    // 7. 清空 io_threads_list[0]
    listEmpty(io_threads_list[0]);

    /* 8. 主线程等待其他 I/O 线程执行完客户端响应 */
    while(1) {
        unsigned long pending = 0;
        for (int j = 1; j < server.io_threads_num; j++)
            pending += getIOPendingCount(j);
        if (pending == 0) break;
    }
 // 9. 全局变量设置为 IO_THREADS_OP_IDLE 表示线程空闲
    io_threads_op = IO_THREADS_OP_IDLE;

    /* 10. 检查 clients_pending_write 中的 client 是否还有数
     要响应给客户端。有的话就调用 CT_Socket.set_write_handler 函数将 sendReplyToClient() 函数设置为 connection-> write_handler 回调函数
    */
    listRewind(server.clients_pending_write,&li);
    while((ln = listNext(&li))) {
        client *c = listNodeValue(ln);

        updateClientMemUsageAndBucket(c);

        if (clientHasPendingReplies(c)) {
            installClientWriteHandler(c);
        }
    }
    // 11. 清空 server.clients_pending_write 队列
    listEmpty(server.clients_pending_write);

    server.stat_io_writes_processed += processed;

    return processed;
}

1. 查询 server.clients_pending_write队列长度，为空则停止后续流程。
2. 如果没有启用 I/O 线程或者只有少量 client 需要处理，主线程同步调用 handleClientsWithPendingWrites() 完成全部 client 的写回响应，不使用 I/O 线程。不满足以上条件，则走接下来的步骤。
3. 开启了 I/O 多线程模式，但是没有激活的话则调用 startThreadedIO 激活 I/O 线程。
4. 循环遍历 clients_pending_write 队列，使用 Round-Robin 算法将 client 分配到每个 I/O 线程绑定的 io_threads_list 列表中。主线程负责 io_threads_list[0] 的任务。
5. 修改全局变量 io_threads_op = IO_THREADS_OP_WRITE，通知 I/O 线程处理的是可写事件。I/O 线程就会执行 writeToClient()函数将 Client 的响应写回客户端。
6. 主线程也会调用 writeToClient() 把 io_threads_list[0] 的 client 执行结果写回给客户端。
7. 主线程执行 listEmpty(io_threads_list[0])清空 io_threads_list[0] 列表。
8. 主线程阻塞等待其他 I/O 线程执行完对应的 io_threads_list 列表中的 Client 执行结果写回客户端。
9. 主线程将全局变量io_threads_op 设置成 IO_THREADS_OP_IDLE，表示 I/O 线程空闲。
10. 检查 clients_pending_write 中的 client 是否还有数据要响应给客户端。有的话就调用 CT_Socket.set_write_handler 函数将 sendReplyToClient() 函数设置为`` connection-> write_handler回调函数。当连接事件变成可写时候，主线程会调用 sendReplyToClient() 函数，内部会调用 writeToClient()` 函数将 client 执行结果数据写回客户端。
11. 调用 listEmpty(server.clients_pending_write)清空 clients_pending_write队列。

一条 Redis 命令的执行过程到此结束，完结撒花。