Redis 是如何提供服务的

在刚刚接触 Redis 的时候，最想要知道的是一个’set name Jhon’ 命令到达 Redis 服务器的时候，它是如何返回’OK’ 的？里面命令处理的流程如何，具体细节怎么样？你一定有问过自己。阅读别人的代码是很枯燥的，但带着好奇心阅读代码，是一件很兴奋的事情，接着翻到了 Redis 源码的 main() 函数。

Redis 在启动做了一些初始化逻辑，比如配置文件读取，数据中心初始化，网络通信模块初始化等，待所有初始化任务完毕后，便开始等待请求。

当请求到来时，Redis 进程会被唤醒，原理是 epoll. select, kqueue 等一些 I/O 多路复用的系统调用。如果有阅读上一章节，应该理解这一句话。接着读取来来自客户端的数据，解析命令，查找命令，并执行命令。

执行命令’set name Jhon’ 的时候，Redis 会在预先初始化好的哈希表里头， 查找 key=’name’ 对应的位置，并存入。

最后，把回复的内容准备好回送给客户端，客户端于是收到了’OK’. 接下来，我们看看详细的过程是怎么样的。

详细的过程

了解了 Redis 的事件驱动模型后，带着命令是如何被处理的这个问题去读代码。刚开始的时候，会有一堆的变量和函数等着读者，但只要抓住主干就好了，下面就是 Redis 的主干部分。

int main(int argc, char **argv) {
   ......
   // 初始化服务器配置，主要是填充 redisServer 结构体中的各种参数
   initServerConfig();
   ......
   // 初始化服务器
   initServer();
   ......
   // 进入事件循环
   aeMain(server.el);
}

分别来看看它们主要做了什么？

initServerConfig()

initServerConfig() 主要是填充 struct redisServer 这个结构体，Redis 所有相关的配置都 在里面。

void initServer() {
   // 创建事件循环结构体
   server.el = aeCreateEventLoop(server.maxclients+REDIS_EVENTLOOP_FDSET_INCR);
   // 分配数据集空间
   server.db = zmalloc(sizeof(redisDb)*server.dbnum);
   /* Open the TCP listening socket for the user commands. */
   // listenToPort() 中有调用listen()
    if (server.port != 0 &&
        listenToPort(server.port,server.ipfd,&server.ipfd_count) == REDIS_ERR)
        exit(1);
......
// 初始化redis 数据集
/* Create the Redis databases, and initialize other internal state. */
for (j = 0; j < server.REDIS_DEFAULT_DBNUM; j++) { // 初始化多个数据库
    // 哈希表，用于存储键值对
    server.db[j].dict = dictCreate(&dbDictType,NULL);
    // 哈希表，用于存储每个键的过期时间
    server.db[j].expires = dictCreate(&keyptrDictType,NULL);
    server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
    server.db[j].ready_keys = dictCreate(&setDictType,NULL);
    server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
    server.db[j].id = j;
    server.db[j].avg_ttl = 0;
  }
......
// 创建接收TCP 或者UNIX 域套接字的事件处理
// TCP
/* Create an event handler for accepting new connections in TCP and Unix
* domain sockets. */
for (j = 0; j < server.ipfd_count; j++) {
    // acceptTcpHandler() tcp 连接接受处理函数
    if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
        acceptTcpHandler,NULL) == AE_ERR)
        {
            redisPanic(
                "Unrecoverable error creating server.ipfd file event.");
        }
   }
......
}

在这里，创建了事件中心，是 Redis 的网络模块，如果你有学过 linux 下的网络编程，那么知道这里一定和 select/epoll/kqueue 相关。

接着，是初始化数据中心，我们平时使用 Redis 设置的键值对，就是存储在里面。这里不急着深入它是怎么做到存储我们的键值对的，接着往下看好了，因为我们主要是想把大致的脉络弄清楚。

在最后一段的代码中，Redis 给 listen fd 注册了回调函数 acceptTcpHandler，也就是说当新的客户端连接的时候，这个函数会被调用，详情接下来再展开。

aeMain()

接着就开始等待请求的到来。

void aeMain(aeEventLoop *eventLoop) {
    eventLoop->stop = 0;
    while (!eventLoop->stop) {
    // 进入事件循环可能会进入睡眠状态。在睡眠之前，执行预设置
    // 的函数aeSetBeforeSleepProc()。
    if (eventLoop->beforesleep != NULL)
        eventLoop->beforesleep(eventLoop);
    // AE_ALL_EVENTS 表示处理所有的事件
    aeProcessEvents(eventLoop, AE_ALL_EVENTS);
  }
}

前面的两个函数都属于是初始化的工作，到这里的时候，Redis 正式进入等待接收请求的状态。具体的实现，和 select/epoll/kqueue 这些 IO 多路复用的系统调用相关，而这也是网络编程的基础部分了。继续跟踪调用链：

int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
    ......
    // 调用IO 多路复用函数阻塞监听
    numevents = aeApiPoll(eventLoop, tvp);
    // 处理已经触发的事件
for (j = 0; j < numevents; j++) {
    // 找到I/O 事件表中存储的数据
    aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
    int mask = eventLoop->fired[j].mask;
    int fd = eventLoop->fired[j].fd;
    int rfired = 0;
/* note the fe->mask & mask & ... code: maybe an already processed
* event removed an element that fired and we still didn't
* processed, so we check if the event is still valid. */
    // 读事件
    if (fe->mask & mask & AE_READABLE) {
        rfired = 1;
        fe->rfileProc(eventLoop,fd,fe->clientData,mask);
    }
    // 写事件
    if (fe->mask & mask & AE_WRITABLE) {
    if (!rfired || fe->wfileProc != fe->rfileProc)
        fe->wfileProc(eventLoop,fd,fe->clientData,mask);
    }
    processed++;
  }
}
// 处理定时事件
/* Check time events */
if (flags & AE_TIME_EVENTS)
    processed += processTimeEvents(eventLoop);
return processed; /* return the number of processed file/time events */
}

可以看到，aeApiPoll 即是 IO 多路复用调用的地方，当有请求到来的时候，进程会觉醒以处理到来的请求。如果你有留意到上面的定时事件处理，也就明白相应的定时处理函数是在哪里触发的了。

新连接的处理流程

在 initServer() 的讲解中，Redis 注册了回调函数 acceptTcpHandler()，当有新的连接到来时，这个函数会被回调，上面的函数指针 rfileProc() 实际上就是指向了 acceptTcpHandler()。

下面是 acceptTcpHandler() 的核心代码：

// 用于TCP 接收请求的处理函数
void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
    int cport, cfd;
    char cip[REDIS_IP_STR_LEN];
    REDIS_NOTUSED(el);
    REDIS_NOTUSED(mask);
    REDIS_NOTUSED(privdata);
    // 接收客户端请求
    cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport);
    // 出错
    if (cfd == AE_ERR) {
        redisLog(REDIS_WARNING,"Accepting client connection: %s", server.neterr);
        return;
  }
    // 记录
    redisLog(REDIS_VERBOSE,"Accepted %s:%d", cip, cport);
    // 真正有意思的地方
    acceptCommonHandler(cfd,0);
}

anetTcpAccept 是接收一个请求cfd，真正有意思的地方是acceptCommonHandler，而 acceptCommonHandler 最核心的调用是 createClient。Redis 对于每一个客户端的连接，都会对应一个结构体 struct redisClient。下面是 createClient 的核心代码：

redisClient *createClient(int fd) {
    redisClient *c = zmalloc(sizeof(redisClient));
    /* passing -1 as fd it is possible to create a non connected client.
    * This is useful since all the Redis commands needs to be executed
    * in the context of a client. When commands are executed in other
    * contexts (for instance a Lua script) we need a non connected client. */
    if (fd != -1) {
        anetNonBlock(NULL,fd);
        anetEnableTcpNoDelay(NULL,fd);
    if (server.tcpkeepalive)
        anetKeepAlive(NULL,fd,server.tcpkeepalive);
    // 为接收到的套接字注册监听事件
    // readQueryFromClient() 应该为处理客户端请求的函数
    if (aeCreateFileEvent(server.el,fd,AE_READABLE,
        readQueryFromClient, c) == AE_ERR)
    {
        close(fd);
        zfree(c);
        return NULL;
    }
  }
  ......
  return c;
}

可以看到，createClient 在事件中心为与客户端连接的套接字注册了 readQueryFrom-Client() 回调函数，而这也就是说当客户端有请求数据过来的时候，acceptTcpHandler() 会被调用。于是，我们找到了’set name Jhon’ 开始处理的地方。

请求的处理流程

readQueryFromClient() 则是获取来自客户端的数据，接下来它会调用processInput-Buffer() 解析命令和执行命令，对于命令的执行，调用的是函数 processCommand()。下面是 processCommand() 核心代码：

int processCommand(redisClient *c) {
    ......
    // 查找命令，redisClient.cmd 在此时赋值
    /* Now lookup the command and check ASAP about trivial error conditions
    * such as wrong arity, bad command name and so forth. */
    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;
        // 参数个数不符合
    } else if ((c->cmd->arity > 0 && c->cmd->arity != c->argc) ||
               (c->argc < c->cmd->arity)) {
        flagTransaction(c);
        addReplyErrorFormat(c,"wrong number of arguments for '%s' command",
            c->cmd->name);
        return REDIS_OK;
}
.....
    // 加入命令队列的情况
    /* Exec the command */
    if (c->flags & REDIS_MULTI &&
        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
        c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
    {
    // 命令入队
    queueMultiCommand(c);
    addReply(c,shared.queued);
    // 真正执行命令。
    // 注意，如果是设置了多命令模式，那么不是直接执行命令，而是让命令入队
    } else {
        call(c,REDIS_CALL_FULL);
    if (listLength(server.ready_keys))
        handleClientsBlockedOnLists();
    }
return REDIS_OK;
}

如上可以看到，Redis 首先根据客户端给出的命令字在命令表中查找对应的 c->cmd, 即 struct redisCommand()。

c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);

Redis 在初始化的时候准备了一个大数组，初始化了所有的命令，即初始化多个 structredisCommand，在 struct redisCommand 中就有该命令对应的回调函数指针。找到命令结构体后，则开始执行命令，核心调用是call()。

执行命令

call() 做的事情有很多，但这里只关注这一句话：call() 调用了命令的回调函数。

    // call() 函数是执行命令的核心函数，真正执行命令的地方
    /* Call() is the core of Redis execution of a command */
void call(redisClient *c, int flags) {
    ......
    // 执行命令对应的处理函数
    c->cmd->proc(c);
    ......
}

对于’set name Jhon’ 命令，对应的回调函数是 setCommand() 函数。setCommand 对 set 命令的参数做了检测，因为还提供设置一个键值对的过期时间等功能，这里只关注最简单的情况。

void setCommand(redisClient *c) {
    ......
    setGenericCommand(c,flags,c->argv[1],c->argv[2],expire,unit,NULL,NULL);
}
void setGenericCommand(redisClient *c, int flags, robj *key,
        robj *val, robj *expire, int unit, robj *ok_reply,
        robj *abort_reply) {
    ......
    setKey(c->db,key,val);
    ......
    addReply(c, ok_reply ? ok_reply : shared.ok);
}
void setKey(redisDb *db, robj *key, robj *val) {
    if (lookupKeyWrite(db,key) == NULL) {
        dbAdd(db,key,val);
    } else {
    dbOverwrite(db,key,val);
  }
  ......
}

setKey() 首先查看 key 是否存在于数据集中，如果存在则覆盖写；如果不存在则添加到数据集中。这里关注 key 不存在的情况：

void dbAdd(redisDb *db, robj *key, robj *val) {
    sds copy = sdsdup(key->ptr);
    int retval = dictAdd(db->dict, copy, val);
    redisAssertWithInfo(NULL,key,retval == REDIS_OK);
}

dictAdd() 就是把 key 存到字典中，实际上即是存到一个哈希表。

在哪里回复客户端

最后，回到 setGenericCommand(), 会调用 addReply()。addReply() 会为与客户端连接的套接字注册可写事件，把’ok’ 添加到客户端的回复缓存中。待再一次回到事件循环的时候，如果这个套接字可写，相应的回调函数就可以被回调了。回复缓存中的数据会被发送到客户端。

由此’set name Jhon’ 命令执行完毕。在把这个流程捋顺的过程，我省去了很多的细节，只关注场景最简单情况最单一的时候，其他的代码都没有去看，譬如主从复制的，持久化的相关逻辑。这对我们快速了解一个系统的原理是很关键的。同样，在面对其他系统代码的时候，也可以带着这三个最简单的问题去阅读：它是谁，它从哪里来，又到哪里去。