redis学习笔记（12）---server基本流程

server工作流程

　　当执行./redis-server后，redis数据库的server端就会启动。

　　然后就会执行redis.c中的main（）函数

　　其中main（）函数中的工作可以主要分为以下几个部分：

　　
1、初始化server端的配置信息- - -initServerConfig()
2、解析运行时的命令参数，并根据参数进行处理，eg：./redis-server - -help
3、如果设置了daemonize参数，则将server设为deamon进程- - -daemonize()
4、启动server- - -initServer()
5、设置周期性处理函数beforeSleep()
6、开始工作- - -aeMain()

1、initServerConfig()

　　初始化server端的配置信息，保存在服务器实例server中，包括监听端口、DB数、命令表等信息。

　　

2、daemonize()　　

　　将进程设为守护进程。

　　守护进程的相关知识之前在linux进程基础 中已经进行了简单介绍。　　

void daemonize(void) {
int fd;
if (fork() != 0) exit(0); /* parent exits */
setsid(); /* create a new session */
if ((fd = open("/dev/null", O_RDWR, 0)) != -1) {
dup2(fd, STDIN_FILENO);
dup2(fd, STDOUT_FILENO);
dup2(fd, STDERR_FILENO);
if (fd > STDERR_FILENO) close(fd);
}
}

3、initServer()

　　这个函数中完成了非常多的任务，包括设置信号处理函数、 创建clients队列、slaves队列、创建数据库、创建共享对象等。

　　除此之外最重要的两个任务是创建监听socket并监听client、以及创建周期性处理事件。

　　我们知道，任何一个服务器的的事件都可以分为IO读写事件和时间处理事件，redis同样如此。

　　1）IO读写事件，包括监听客户端的连接以及与客户端进行数据交互等

　　2）时间处理事件，在设定的时间处理相关事件，包括周期性刷新数据库等。

　　这两类事件都是通过server.el这个变量来保存的。

3.1、IO读写事件

　　首先redis会为服务器创建一个监听fd，来监听来自客户端的连接，主要会调用到以下两个函数　　

listenToPort(server.port,server.ipfd,&server.ipfd_count);
aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE, acceptTcpHandler,NULL);

listenToPort：根据传入的port创建监听描述符，同时调用fcntl()将fd设为非阻塞的，然后调用bind()、listen()函数。注意redis会分别根据IPv4、IPv6两种地址分别创建一个socket

aeCreateFileEvent：创建读写事件。对于监听描述符而言，只需要创建一个读事件监听来自client的连接即可。注意，监听描述符的回调函数为acceptTcpHandler

　　最后将该事件加入到server.el结构中

3.2、时间处理事件

　　对于server端，redis会周期性的执行serverCron（）来完成一些处理，因此将这个事件也加入到server.el结构中　　

aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
//每1ms执行一次serverCron()

4、beforeSleep()

　　在redis的事件主循环中，每次循环都会执行一次，其中包括向所有slave发送ACK、写AOF文件等操作

5、aeMain()　　

　　redis服务器端最重要的函数，为redis的事件主循环。如果redis没有接收到中断信号，那么就会一直循环执行这个函数。　　

aeMain(server.el);
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
while (!eventLoop->stop) {
if (eventLoop->beforesleep != NULL)
eventLoop->beforesleep(eventLoop);
aeProcessEvents(eventLoop, AE_ALL_EVENTS);
}
}

　　可以发现，该函数就是死循环的执行beforesleep()和aeProcessEvents()。

　　作为redis服务器端的核心流程，aeProcessEvents()的实现代码较长，但是主要也只有3个动作

　　
1、计算调用select、epoll等函数可以阻塞的时间
2、调用aeApiPoll()等待IO事件发生，若有事件发生，则调用相应的回调函数
3、调用processTimeEvents()处理时间事件

　　

　　由于select、epoll等IO复用机制在一定时间内没有事件发生时，会一直阻塞在那里。因此为了不影响后面时间事件的处理，必须在最近的一个时间事件到来之前，完成IO复用机制的调用。因此首先找到最近一个时间事件，计算距离当前时间的时间差，来作为调用aeApiPoll()的参数。　　

int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
if (eventLoop->maxfd != -1 ||
((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
//1、有时间事件时，计算时间差
if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
shortest = aeSearchNearestTimer(eventLoop); //找到最近一个时间事件
if (shortest) {
aeGetTime(&now_sec, &now_ms); //得到当前时间
tvp = &tv;
//计算时间差tvp
tvp->tv_sec = shortest->when_sec - now_sec;
if (shortest->when_ms < now_ms) {
tvp->tv_usec = ((shortest->when_ms+1000) - now_ms)*1000;
tvp->tv_sec --;
} else {
tvp->tv_usec = (shortest->when_ms - now_ms)*1000;
}
if (tvp->tv_sec < 0) tvp->tv_sec = 0;
if (tvp->tv_usec < 0) tvp->tv_usec = 0;
} else {
if (flags & AE_DONT_WAIT) {  //此时不阻塞，立即返回
tv.tv_sec = tv.tv_usec = 0;
tvp = &tv;
} else { //否则可以永远等待
tvp = NULL; /* wait forever */
}
}
//2、调用IO复用机制，处理IO事件
numevents = aeApiPoll(eventLoop, tvp);
for (j = 0; j < numevents; j++) {
aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
int mask = eventLoop->fired[j].mask;
int fd = eventLoop->fired[j].fd;
int rfired = 0;
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++;
}
}
//3、处理时间事件
if (flags & AE_TIME_EVENTS)
processed += processTimeEvents(eventLoop);
return processed;
}

　　这就是redis服务器侧主要的工作流程了。

具体例子：

1、当有一个client连接到来时

　　此时redis服务器端的监听描述符就会有事件发生，之前已经提到过该fd上只注册了读事件acceptTcpHandler()，因此执行

fe->rfileProc(eventLoop,fd,fe->clientData,mask);

就会调用acceptTcpHandler()函数

　　1）首先acceptTcpHandler()会调用accept获取连接描述符cfd

　　2）然后调用acceptCommonHandler()创建一个client实例　　

　　3）在createClient()中会为每个连接描述符注册读事件readQueryFromClient() ，同时将每个client的默认数据库设为0

　　这样client和server端的连接就建立好了。当有客户端请求到来时，就会执行readQueryFromClient()函数 ，来处理该请求了。

void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
while(max--) {
cfd = anetTcpAccept(server.neterr, fd, cip, sizeof(cip), &cport); //1、调用accept获取连接描述符cfd
if (cfd == ANET_ERR) {
......
}
acceptCommonHandler(cfd,0);
}
}
static void acceptCommonHandler(int fd, int flags) {
redisClient *c;
if ((c = createClient(fd)) == NULL) { //2、创建client实例
close(fd);
return;
}
}
redisClient *createClient(int fd) {
redisClient *c = zmalloc(sizeof(redisClient));
if (fd != -1) {
anetNonBlock(NULL,fd);
if (aeCreateFileEvent(server.el,fd,AE_READABLE,
readQueryFromClient, c) == AE_ERR){  //3、注册读事件
/*  ......  */
}
}
selectDb(c,0);
/*  ......  */
return c;
}

2、当有客户端请求到来时（eg：执行命令 set key value ）

2.1、readQueryFromClient()读入请求并处理

　　首先连接fd上的读事件会被触发，因此server端会调用readQueryFromClient()来进行处理。主要过程是：

　　　　
1、为接收缓存区申请内存
2、调用read从客户端读入请求数据到c->querybuf中
3、调用processInputBuffer对请求进行处理

　　

void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
c->querybuf = sdsMakeRoomFor(c->querybuf, readlen);  //1、申请空间
nread = read(fd, c->querybuf+qblen, readlen);  //2、读请求
processInputBuffer(c);   //3、处理输入请求
}

　　当执行命令 set key value后，打印c->querybuf得到如下结果：

　　


　　即其中内容的格式为：

　　

2.2、processInputBuffer()处理请求

　　在processInputBuffer()中

　　1）首先调用processMultibulkBuffer，按照协议格式，将接收缓冲区中的内容解析出来，并为每个参数创建一个字符串对象robject 　

//主要处理如下
ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll); //1、首先解析出参数个数并转换成数字
c->multibulklen = ll; //将参数个数赋给multibulklen
c->argv = zmalloc(sizeof(robj*)*c->multibulklen);  //2、为对象分配内存
c->argv[c->argc++] = createStringObject(c->querybuf+pos,c->bulklen); //3、依次创建ll个对象

　　对于本例，ll = 3，最终会生成3个字符串对象，字符串的内容分别为”set” 、”key” 、”value”。

　　c->argv的类型为

robj **argv;

，因此可以将其看作一个数组，数组中的每一项指向一个robj。

　　在上一章已经讲过，当字符串长度小于39字节时，会采用embstr编码方式来组织数据，因此最终c->argv的内容如下：

　　


　　　　

　　2）调用processCommand对命令进行处理

　　首先查找命令，当命令不存在或参数个数不对时，错误则直接返回。

　　然后中间会进行一系列判断，暂时不管

　　最后就调用call()处理命令

　　如本例中的命令为set，因此会在redisCommandTable中找到set命令，然后执行set命令对应的函数setCommand()

int processCommand(redisClient *c) {
c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
if (!c->cmd) {  //没有找到命令
return REDIS_OK;
} else if ((c->cmd->arity > 0 && c->cmd->arity != c->argc) ||
(c->argc < -c->cmd->arity)) { //命令参数个数错误
return REDIS_OK;
}
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);  //执行命令
}
return REDIS_OK;
}

　　3）调用setCommand()执行命令

　　setCommand()在上一篇中已经进行了简单的介绍，需要注意的是最后setCommand()会执行

　　

addReply(c, ok_reply ? ok_reply : shared.ok);

将操作的结果返回给客户端

　　

2.3、调用addReply将结果返回给client

　　1）该函数会调用prepareClientToWrite()首先将要返回给客户端的结果按照一定的格式保存到缓冲区中

　　2）然后调用

aeCreateFileEvent(server.el, c->fd, AE_WRITABLE,sendReplyToClient, c)

在该连接fd上注册写事件。

　　这样当server下一次执行aeMain函数时，就会检测到有写事件发生，就会调用sendReplyToClient()函数了。

　　3）在sendReplyToClient()函数中，就会调用write系统调用将结果通过socket返回给client了。

　　这样整个set key value命令就执行完了

本文所引用的源码全部来自Redis3.0.7版本

redis学习参考资料：

https://github.com/huangz1990/redis-3.0-annotated

Redis 设计与实现（第二版）