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

上一篇: Redis 事件驱动详...下一篇: Redis 数据结构综...

• 被带到火星去了

只有登录了才能参与评论，快登录！如果你还没有账号你可以注册一个账号。