在刚刚接触 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 数据结构综...
被带到火星去了
只有登录了才能参与评论，快登录！如果你还没有账号你可以注册一个账号。
本文暂时没有评论，来添加一个吧(●'◡'●)