【Redis源码学习】浅析redis中的IO多路复用机制

妖狐艹你老母 2022-12-04 09:17 77阅读 0赞

我们知道，我们在使用redis的时候，通过客户端发送一个get命令，就能够得到redis服务端返回的数据。redis是基于传统的C/S架构实现的。它通过监听一个TCP端口（6379）的方式来接收来自客户端的连接，从而进行后续命令的执行，并把执行结果返回给客户端。

redis是一个合格的服务端程序  
我们先思考一个问题：作为一个合格的服务端程序，我们在命令行输入一个get命令之后，redis服务端是怎么处理这个命令，并把结果返回给客户端的呢？

客户端与服务器需要分别创建一个套接字表明自己所在的网络地址与端口号，然后基于TCP协议来进行套接字之间的通信。通常情况下，一个服务端程序的socket通信流程如下：

int main(int argc, char *argv[]) { 
        listenSocket = socket(); //调用socket()系统调用创建一个监听套接字描述符
        bind(listenSocket);  //绑定地址与端口
        listen(listenSocket); //由默认的主动套接字转换为服务器适用的被动套接字
        while (1) {  //不断循环去监听是否有客户端连接事件到来
            connSocket = accept($listenSocket); //接受客户端连接
            read(connsocket); //从客户端读取数据，只能同时处理一个客户端
            write(connsocket); //返回给客户端数据，只能同时处理一个客户端
        }
        return 0;
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTg5NzY5_size_16_color_FFFFFF_t_70_pic_center]

**Redis源码剖析训练营内容：**  
**1：数据存储分析  
2：redis存储原理  
3：redis事件机制  
4：redis6.0-IO多线程  
5：redis有序集合的实现-跳表  
6：跳表的应用以及与其他动态有序搜索结构比较  
7：redis-driver实现  
8：redis高可用部署策略  
redis源码剖析训练营学习地址:https://ke.qq.com/course/2993468?flowToken=1027484**

在redis中，同样要经过以上几个步骤。与客户端建立连接之后，就会读取客户端发来的命令，然后执行命令，最后通过调用write系统调用，将命令的执行结果返回给客户端。

但是这样一个进程只能同时处理一个客户端的连接与读写事件。为了让单进程的服务端应用同时处理多个客户端的事件，我们采用了IO多路复用机制。目前最好的IO多路复用机制就是epoll。使用epoll创建的服务器代码：

int main(int argc, char *argv[]) { 
    
        listenSocket = socket(AF_INET, SOCK_STREAM, 0); //同上，创建一个监听套接字描述符
        
        bind(listenSocket)  //同上，绑定地址与端口
        
        listen(listenSocket) //同上，由默认的主动套接字转换为服务器适用的被动套接字
        
        epfd = epoll_create(EPOLL_SIZE); //创建一个epoll实例
        
        ep_events = (epoll_event*)malloc(sizeof(epoll_event) * EPOLL_SIZE); //创建一个epoll_event结构存储套接字集合
        event.events = EPOLLIN;
        event.data.fd = listenSocket;
        
        epoll_ctl(epfd, EPOLL_CTL_ADD, listenSocket, &event); //将监听套接字加入到监听列表中
        
        while (1) { 
        
            event_cnt = epoll_wait(epfd, ep_events, EPOLL_SIZE, -1); //等待返回已经就绪的套接字描述符们
            
            for (int i = 0; i < event_cnt; ++i) {  //遍历所有就绪的套接字描述符
                if (ep_events[i].data.fd == listenSocket) {  //如果是监听套接字描述符就绪了，说明有一个新客户端连接到来
                
                    connSocket = accept(listenSocket); //调用accept()建立连接
                    
                    event.events = EPOLLIN;
                    event.data.fd = connSocket;
                    
                    epoll_ctl(epfd, EPOLL_CTL_ADD, connSocket, &event); //添加对新建立的连接套接字描述符的监听，以监听后续在连接描述符上的读写事件
                    
                } else {  //如果是连接套接字描述符事件就绪，则可以进行读写
                
                    strlen = read(ep_events[i].data.fd, buf, BUF_SIZE); //从连接套接字描述符中读取数据, 此时一定会读到数据，不会产生阻塞
                    if (strlen == 0) {  //已经无法从连接套接字中读到数据，需要移除对该socket的监听
                    
                        epoll_ctl(epfd, EPOLL_CTL_DEL, ep_events[i].data.fd, NULL); //删除对这个描述符的监听
                        
                        close(ep_events[i].data.fd);
                    } else { 
                        write(ep_events[i].data.fd, buf, str_len); //如果该客户端可写 把数据写回到客户端
                    }
                }
            }
        }
        close(listenSocket);
        close(epfd);
        return 0;
    }

redis基于原有的select、poll与epoll机制，结合自己独特的业务需求，封装了自己的一套事件处理函数，我们把它叫做ae（a simple event-driven programming library）。而redis具体使用select、epoll还是mac上的kqueue技术，redis会首先进行判断，然后选择性能最优的那个：

/* Include the best multiplexing layer supported by this system. * The following should be ordered by performances, descending. */
    #ifdef HAVE_EVPORT
    #include "ae_evport.c"
    #else
        #ifdef HAVE_EPOLL
        #include "ae_epoll.c"
        #else
            #ifdef HAVE_KQUEUE
            #include "ae_kqueue.c"
            #else
            #include "ae_select.c"
            #endif
        #endif
    #endif

因为 select 函数是作为 POSIX 标准中的系统调用，在不同版本的操作系统上都会实现，所以将其作为兜底方案。

redis中的IO多路复用  
当我们在命令行中启动一个redis-server的时候，redis其实做了和我们之前写的epoll服务器类似的操作，重点的函数调用有以下三个：

int main(int argc, char **argv) { 
        ...
        initServerConfig(); //初始化存储服务端信息的结构体
        ...
        initServer(); //初始化redis事件循环并调用epoll_create与epoll_ctl。创建socket、bind、listen、accept都在这个函数中进行调用，并注册调用后返回的监听描述符和连接描述符
        ...
        aeMain(); //执行while(1)事件循环，并调用epoll_wait获取已就绪的描述符，并调用对应的handler
        ...
    }

接下来我么一个一个来看：

initServerConfig()  
redis服务端的所有信息都存储在一个redisServer结构体中，这个结构体字段非常多，比如服务端的套接字信息（如地址和端口），还有很多支持redis其他功能如集群、持久化等的配置信息都存储在这个结构体中。这个函数调用就是对redisServer结构体的所有字段进行初始化并赋一个初始值。由于我们这次讲解的是事件与IO多路复用机制在redis中的应用，所以我们只关注其中的几个字段即可。

initServer()  
这个函数调用是我们的重中之重。初始化完服务器的相关信息之后，就需要进行套接字的创建、绑定、监听并与客户端建立连接了。在这个函数中，进行了我们常说的创建socket、bind、listen、accept、epoll\_create、epoll\_ctl调用，我们可以对照上文的epoll服务器，逐步了解redis的事件机制。initServer()的主要函数调用如下：

void initServer(void) { 
        ...
        server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR); 
        ...
    
        if (server.port != 0 && listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
            exit(1);
        ...
    
        for (j = 0; j < server.ipfd_count; j++) { 
            if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE, acceptTcpHandler,NULL) == AE_ERR){ 
                    serverPanic(
                        "Unrecoverable error creating server.ipfd file event.");
                }
        }
        ...
    }

我们按照从上到下的顺序解读这几行关键代码，以下均以使用epoll多路复用机制为例：

aeCreateEventLoop()  
在redis中，有一个aeEventLoop的概念，它来管理所有相关的事件描述字段、存储已注册的事件、已就绪的事件：

typedef struct aeEventLoop { 
        int stop; //标识事件循环（即while(1)）是否结束
        
        aeFileEvent *events;  //存储已经注册的文件事件（文件事件即客户端连接与读写事件）
        aeFiredEvent *fired;  //存储已就绪的文件事件
        aeTimeEvent *timeEventHead; //存储时间事件（时间事件后面再讲）
        
        void *apidata; /* 存储epoll相关信息 */
        
        aeBeforeSleepProc *beforesleep; //事件发生前需要调用的函数
        aeBeforeSleepProc *aftersleep; //事件发生后需要调用的函数
    } aeEventLoop;

redis将所有通过epoll\_wait()返回的就绪描述符都存储在fired数组中，然后遍历这个数组，并调用对应的事件处理函数，一次性处理完所有事件。在aeCreateEventLoop()函数中，对这个管理所有事件信息的结构体字段进行了初始化，这里面也包括调用epoll\_create()，对epoll的epfd进行初始化：

aeEventLoop *aeCreateEventLoop(int setsize) { 
        aeEventLoop *eventLoop;
        int i;
    
        if ((eventLoop = zmalloc(sizeof(*eventLoop))) == NULL) goto err;
        eventLoop->events = zmalloc(sizeof(aeFileEvent)*setsize);
        eventLoop->fired = zmalloc(sizeof(aeFiredEvent)*setsize);
        if (eventLoop->events == NULL || eventLoop->fired == NULL) goto err;
        eventLoop->setsize = setsize;
        eventLoop->lastTime = time(NULL);
        eventLoop->timeEventHead = NULL;
        eventLoop->timeEventNextId = 0;
        eventLoop->stop = 0;
        eventLoop->maxfd = -1;
        eventLoop->beforesleep = NULL;
        eventLoop->aftersleep = NULL;
        if (aeApiCreate(eventLoop) == -1) goto err; //调用aeApiCreate()，内部会调用epoll_create()
        for (i = 0; i < setsize; i++)
            eventLoop->events[i].mask = AE_NONE;
        return eventLoop;
    }

在aeApiCreate()函数中，调用了epoll\_create()，并将创建好的epfd放到eventLoop结构体的apidata字段保管：

typedef struct aeApiState { 
        int epfd;
        struct epoll_event *events;
    } aeApiState;
    
    static int aeApiCreate(aeEventLoop *eventLoop) { 
        aeApiState *state = zmalloc(sizeof(aeApiState));
    
        if (!state) return -1;
        state->events = zmalloc(sizeof(struct epoll_event)*eventLoop->setsize);
        if (!state->events) { 
            zfree(state);
            return -1;
        }
        state->epfd = epoll_create(1024); /* 调用epoll_create初始化epoll的epfd */
        if (state->epfd == -1) { 
            zfree(state->events);
            zfree(state);
            return -1;
        }
        eventLoop->apidata = state; //将创建好的epfd放到eventLoop结构体的apidata字段保管
        return 0;
    }

listenToPort()  
在创建完epfd之后，我们就要进行socket创建、绑定、监听的操作了，这几步在listenToPort()函数来进行：

int listenToPort(int port, int *fds, int *count) { 
        if (server.bindaddr_count == 0) server.bindaddr[0] = NULL;
        for (j = 0; j < server.bindaddr_count || j == 0; j++) {  //遍历所有的ip地址
            if (server.bindaddr[j] == NULL) {  //还没有绑定地址
               ...
            } else if (strchr(server.bindaddr[j],':')) {  //绑定IPv6地址
                ...
            } else {  //绑定IPv4地址，一般会进到这个if分支中
                fds[*count] = anetTcpServer(server.neterr,port,server.bindaddr[j], server.tcp_backlog);  //真正的绑定逻辑
            }
            ...
        }
        return C_OK;
    }

redis会先进行绑定ip地址类型的判断，我们一般是IPv4，所以一般会走到第三个分支，调用anetTcpServer()函数来进行具体的绑定逻辑：

static int _anetTcpServer(char *err, int port, char *bindaddr, int af, int backlog)
    { 
       ...
        if ((rv = getaddrinfo(bindaddr,_port,&hints,&servinfo)) != 0) { 
            anetSetError(err, "%s", gai_strerror(rv));
            return ANET_ERR;
        }
        for (p = servinfo; p != NULL; p = p->ai_next) { 
            if ((s = socket(p->ai_family,p->ai_socktype,p->ai_protocol)) == -1) //调用socket()创建一个监听套接字
                continue;
    
            if (af == AF_INET6 && anetV6Only(err,s) == ANET_ERR) goto error;
            if (anetSetReuseAddr(err,s) == ANET_ERR) goto error;
            if (anetListen(err,s,p->ai_addr,p->ai_addrlen,backlog) == ANET_ERR) s = ANET_ERR; //调用bind()与listen()绑定端口并转化为服务端被动套接字
            goto end;
        }
    }

在调用socket()系统调用创建了套接字之后，需要进一步调用bind()与listen()，这两步是在anetListen()函数内部实现的：

static int anetListen(char *err, int s, struct sockaddr *sa, socklen_t len, int backlog) { 
        if (bind(s,sa,len) == -1) {  //调用bind()绑定端口
            anetSetError(err, "bind: %s", strerror(errno));
            close(s);
            return ANET_ERR;
        }
    
        if (listen(s, backlog) == -1) {  //调用listen()将主动套接字转换为被动监听套接字
            anetSetError(err, "listen: %s", strerror(errno));
            close(s);
            return ANET_ERR;
        }
        return ANET_OK;
    }

看到这里，我们知道redis和epoll服务器一样，都是需要进行套接字创建、绑定、监听的过程。

aeCreateFileEvent  
在redis中，把客户端连接事件、读写事件统称为文件事件。我们刚才完成了socket创建、bind、listen的过程。目前我们已经有了一个监听描述符，那么我们需要首先将监听描述符添加到epoll的监听列表，以监听客户端的连接事件。在initServer()中，通过调用aeCreateFileEvent()，同时指定了它的事件处理函数acceptTcpHandler()来实现对客户端连接事件的处理：

for (j = 0; j < server.ipfd_count; j++) { 
            if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE, acceptTcpHandler,NULL) == AE_ERR){ 
                    serverPanic("Unrecoverable error creating server.ipfd file event.");
            }
        }

跟进aeCreateFileEvent()函数，发现其内部进一步调用了aeApiAddEvent()函数：

int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask, aeFileProc *proc, void *clientData) { 
        if (fd >= eventLoop->setsize) { 
            errno = ERANGE;
            return AE_ERR;
        }
        aeFileEvent *fe = &eventLoop->events[fd];
    
        if (aeApiAddEvent(eventLoop, fd, mask) == -1)
            return AE_ERR;
        fe->mask |= mask;
        if (mask & AE_READABLE) fe->rfileProc = proc;
        if (mask & AE_WRITABLE) fe->wfileProc = proc;
        fe->clientData = clientData;
        if (fd > eventLoop->maxfd)
            eventLoop->maxfd = fd;
        return AE_OK;
    }
    static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) { 
        aeApiState *state = eventLoop->apidata;
        struct epoll_event ee = { 0}; 
        int op = eventLoop->events[fd].mask == AE_NONE ?
                EPOLL_CTL_ADD : EPOLL_CTL_MOD;
    
        ee.events = 0;
        mask |= eventLoop->events[fd].mask;
        if (mask & AE_READABLE) ee.events |= EPOLLIN;
        if (mask & AE_WRITABLE) ee.events |= EPOLLOUT;
        ee.data.fd = fd;
        if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1; //调用epoll_ctl添加客户端连接事件
        return 0;
    }

aeApiAddEvent函数会调用epoll\_ctl()，将客户端连接事件添加到监听列表。同时，redis会将该事件的处理函数放到aeFileEvent结构体中进行存储：

typedef struct aeFileEvent { 
        int mask; /* one of AE_(READABLE|WRITABLE|BARRIER) */
        aeFileProc *rfileProc; //读事件处理程序
        aeFileProc *wfileProc; //写事件处理程序
        void *clientData;  //客户端数据
    } aeFileEvent;

对照之前我们写过的epoll服务端程序，我们已经实现了以下几个步骤：

int main(int argc, char *argv[]) { 
    
        listenSocket = socket(AF_INET, SOCK_STREAM, 0); //创建一个监听套接字描述符
        
        bind(listenSocket)  //绑定地址与端口
        
        listen(listenSocket) //由默认的主动套接字转换为服务器适用的被动套接字
        
        epfd = epoll_create(EPOLL_SIZE); //创建一个epoll实例
        
        ep_events = (epoll_event*)malloc(sizeof(epoll_event) * EPOLL_SIZE); //创建一个epoll_event结构存储套接字集合
        event.events = EPOLLIN;
        event.data.fd = listenSocket;
        
        epoll_ctl(epfd, EPOLL_CTL_ADD, listenSocket, &event); //将监听套接字加入到监听列表中
       ...
    }

我们已经实现了对套接字的创建、bind、listen，已通过epoll\_create()实现了epfd的创建，并将初始的监听套接字描述符事件添加到了epoll的监听列表中，并为他指定了事件处理函数。下一步，就应该到了while(1)循环调用epoll\_wait()的阶段了。通过阻塞调用epoll\_wait()，返回所有已经就绪的套接字描述符，触发相应事件，然后对事件进行处理。

aeMain()  
最后就是通过while(1)循环，等待客户端连接事件的到来啦：

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

在eventLoop中，采用stop标志来判定循环是否结束。如果没有结束，那么循环调用aeProcessEvents()。我们猜测，这里面就调用了epoll\_wait()，阻塞等待事件的到来，然后遍历所有就绪的套接字描述符，然后调用对应的事件处理函数即可：

int aeProcessEvents(aeEventLoop *eventLoop, int flags)
    { 
            numevents = aeApiPoll(eventLoop, tvp); //调用epoll_wait()
            ...
    }

我们跟进aeApiPoll，来看看epoll\_wait()是如何调用的：

static int aeApiPoll(aeEventLoop *eventLoop, struct timeval *tvp) { 
        aeApiState *state = eventLoop->apidata; //
        int retval, numevents = 0;
    
        retval = epoll_wait(state->epfd,state->events,eventLoop->setsize, tvp ? (tvp->tv_sec*1000 + tvp->tv_usec/1000) : -1);
        if (retval > 0) { 
            int j;
            numevents = retval;
            for (j = 0; j < numevents; j++) { 
                int mask = 0;
                struct epoll_event *e = state->events+j;
    
                if (e->events & EPOLLIN) mask |= AE_READABLE;
                if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
                if (e->events & EPOLLERR) mask |= AE_WRITABLE;
                if (e->events & EPOLLHUP) mask |= AE_WRITABLE;
                eventLoop->fired[j].fd = e->data.fd;
                eventLoop->fired[j].mask = mask;
            }
        }
        return numevents;
    }

首先从eventLoop中拿出之前在aeApiCreate()中创建的epfd与已经注册的事件集合，调用epoll\_wait()等待事件们的到来，并返回所有就绪事件的描述符集合。随后，遍历所有就绪的描述符集合，判断它是什么类型的描述符，是可读还是可写的，随后将所有就绪可处理的事件存储到eventLoop中的fired数组中，并把相应数组位置上的可读还是可写标记也一并存储。

回到外部调用处，我们现在已经把所有能够处理的事件都放到了fired数组中，那么我们就可以通过遍历这个数组，拿到所有可以处理的事件，然后调用对应的事件处理函数：

```handlebars
    
    ```handlebars
    int aeProcessEvents(aeEventLoop *eventLoop, int flags)
    { 
            numevents = aeApiPoll(eventLoop, tvp); //调用epoll_wait()
    
            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 fired = 0; 
    
                if (!invert && fe->mask & mask & AE_READABLE) { 
                    fe->rfileProc(eventLoop,fd,fe->clientData,mask); //如果该事件是读事件，调用读事件处理函数
                    fired++;
                }
    
                if (fe->mask & mask & AE_WRITABLE) { 
                    if (!fired || fe->wfileProc != fe->rfileProc) { 
                        fe->wfileProc(eventLoop,fd,fe->clientData,mask); //如果该事件是写事件，调用写事件处理函数
                        fired++;
                    }
                }
            }
        }
        ...
    }

至于如何区分是客户端连接事件以及读写事件，redis通过指定不同的事件处理函数（如accept事件是acceptTcpHandler事件处理函数），读或写事件又是其他的事件处理函数。通过这层封装，免去了判断套接字描述符类型的步骤，直接调用之前注册的事件处理函数即可。

回顾上面的epoll服务器代码，是不是和这一段代码很相似呢？

while (1) { 
        
            event_cnt = epoll_wait(epfd, ep_events, EPOLL_SIZE, -1); //等待返回已经就绪的套接字描述符们
            
            for (int i = 0; i < event_cnt; ++i) {  //遍历所有就绪的套接字描述符
                if (ep_events[i].data.fd == listenSocket) {  //如果是监听套接字描述符就绪了，说明有一个新客户端连接到来
                
                    connSocket = accept(listenSocket); //调用accept()建立连接
                    
                    event.events = EPOLLIN;
                    event.data.fd = connSocket;
                
    
    
                    epoll_ctl(epfd, EPOLL_CTL_ADD, connSocket, &event); //添加对新建立的连接套接字描述符的监听，以监听后续在连接描述符上的读写事件
                    
                } else {  //如果是连接套接字描述符事件就绪，则可以进行读写
                
                    strlen = read(ep_events[i].data.fd, buf, BUF_SIZE); //从连接套接字描述符中读取数据, 此时一定会读到数据，不会产生阻塞
                    if (strlen == 0) {  //已经无法从连接套接字中读到数据，需要移除对该socket的监听
                    
                        epoll_ctl(epfd, EPOLL_CTL_DEL, ep_events[i].data.fd, NULL); //删除对这个描述符的监听
                        
                        close(ep_events[i].data.fd);
                    } else { 
                        write(ep_events[i].data.fd, buf, str_len); //如果该客户端可写 把数据写回到客户端
                    }
                }
            }
        }

至此，我们就掌握了redis中的IO多路复用场景。redis把所有连接与读写事件、还有我们没提到的时间事件一起集中管理，并对底层IO多路复用机制进行了封装，最终实现了单进程能够处理多个连接以及读写事件。这就是IO多路复用在redis中的应用。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTg5NzY5_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/69ea94c286a04824b8dca778d94b4453.png