文章目录
- 1、简述
- 2、多路复用的三个函数
- 3、创建epoll实例
- 4、绑定端口、监听端口
- 5、向epoll实例注册连接事件
- 6、从epoll实例中获取就绪的事件
1、简述
众所周知，redis是一款抗高并发的利器，据官方压测，单机可达10万qps。但背后实际处理命令的线程只有一条，这听上去其实挺匪夷所思的，因为在我们的日常开发中，说到高并发，多线程是一个非常常用的解决方案。那redis凭什么靠一条线程，就能支持高并发呢？最主要的原因，就是标题所说的IO多路复用，IO多路复用是怎么做的呢？这是老八股了，IO多路复用，背后依赖的是多路复用的函数，有select、poll、epoll，linux默认使用的是epoll函数，redis把客户端连接通过epoll函数给到内核，内核监听到连接有可读写的事件，就将该事件返回redis进行处理。那具体的实现细节呢？redis怎么给的内核，内核又怎么返回的?
2、多路复用的三个函数
epoll函数由3个函数组合来完成多路复用这件事。分别是：
epoll_create、epoll_ctl、epoll_wait
1)、epoll_create：创建epoll实例
2)、epoll_ctl：将连接对应的socket描述符注册到epoll实例中
3)、epoll_wait：获取epoll实例中可读写的描述符
画一个简单的流程图串一下这三个函数的作用
从图中可以看出，redis在启动的时候，先是通过epoll_create函数创建epoll实例，然后绑定端口、监听端口，然后通过epoll_ctl函数注册连接事件，最后会搞一个死循环，通过epoll_wait函数获取可读写的事件（每一个事件对应的都是一个可读写的客户端连接）
铺垫完上面的流程，我们看一下源码。redis的启动源码在server.c文件的main方法中，main方法是redis启动的入口，其中有很多流程，我们只关注我们这个流程会用到的函数。
3、创建epoll实例
首先是通过内核提供的epoll_create函数创建epoll实例，这个流程入口在initServer方法中.
void initServer(void) {
......
//创建epoll实例
server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
if (server.el == NULL) {
serverLog(LL_WARNING,
"Failed creating the event loop. Error message: '%s'",
strerror(errno));
exit(1);
}
......
}
aeCreateEventLoop是创建epoll实例的入口，我们进入这个方法。
aeEventLoop *aeCreateEventLoop(int setsize) {
......
if (aeApiCreate(eventLoop) == -1) goto err;
......
}
其中又调用了一个aeApiCreate方法，这个方法是对epoll_create函数做了一层封装，我们继续进入aeApiCreate方法。
static int aeApiCreate(aeEventLoop *eventLoop) {
......
//创建epoll实例
//这里的1024并不是说epoll函数只能监听1024个描述符.因为在2.6.8内核之后,内核维护的是一个动态的队列,理论上我们可以一直添加描述符
state->epfd = epoll_create(1024); /* 1024 is just a hint for the kernel */
......
}
4、绑定端口、监听端口
这里就看到了我们想找的epoll_create函数。
创建完epoll实例后，接下来就是绑定端口、监听端口。
这部分的代码也是在initServer方法中，就在创建epoll实例的下方
void initServer(void) {
......
//创建epoll实例
server.el = aeCreateEventLoop(server.maxclients+CONFIG_FDSET_INCR);
if (server.el == NULL) {
serverLog(LL_WARNING,
"Failed creating the event loop. Error message: '%s'",
strerror(errno));
exit(1);
}
server.db = zmalloc(sizeof(redisDb)*server.dbnum);
......
//绑定、监听端口
if (server.port != 0 &&
listenToPort(server.port,server.ipfd,&server.ipfd_count) == C_ERR)
exit(1);
......
}
绑定、监听端口的逻辑在listenToPort方法中，该方法的入参有3个值，第一个就是要绑定、监听的端口，默认是6379。第二个值是描述符，第三个是描述符的数量。这个时候，后面这两个参数还没值，需要到listenToPort方法中赋值。
int listenToPort(int port, int *fds, int *count) {
......
//绑定IPV6
fds[*count] = anetTcp6Server(server.neterr,port,NULL,server.tcp_backlog);
......
//绑定IPV4
fds[*count] = anetTcpServer(server.neterr,port,NULL,server.tcp_backlog);
......
(*count)++;
}
所以最终fds数组一共赋值2个值。count赋值2
5、向epoll实例注册连接事件
这个逻辑还是在initServer方法中。server.ipfd_count的值就是上面的那个count值，是2。所以这个循环会执行2次，注册2个连接事件，一个IPV4、一个IPV6
aeCreateFileEvent，是一个非常重要的方法，是用来创建事件的。该方法有5个入参。
第一个是redis对应epoll实例的结构体，第二个是需要注册的描述符，第三个是需要注册的事件类型，第四个是事件触发后的回调函数，最后一个是客户端数据。我们是注册连接事件，所以不会有客户端数据，此时客户端还没有连接redis
void initServer(void) {
......
//注册连接事件
for (j = 0; j < server.ipfd_count; j++) {
......
if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,acceptTcpHandler,NULL) == AE_ERR)
......
}
......
}
我们进入aeCreateFileEvent方法，
int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
aeFileProc *proc, void *clientData)
{
......
//aeApiAddEvent函数内部调用epoll_ctl函数
if (aeApiAddEvent(eventLoop, fd, mask) == -1)
return AE_ERR;
......
//将acceptTcpHandler回调函数挂到当前连接事件上
if (mask & AE_READABLE) fe->rfileProc = proc;
if (mask & AE_WRITABLE) fe->wfileProc = proc;
......
}
aeCreateFileEvent主要就是做两件事，注册连接事件、给事件挂回调函数，回调函数就是acceptTcpHandler。aeApiAddEvent是对epoll_ctl函数的封装。我们进入其中看一下
static int aeApiAddEvent(aeEventLoop *eventLoop, int fd, int mask) {
......
//调用epoll的epoll_ctl函数注册事件,一共4个参数。
//1、epoll实例
//2、要执行的操作类型,添加事件还是修改修改事件。第一次肯定是添加事件
//3、要监听的文件描述符
//4、epoll_event类型变量
if (epoll_ctl(state->epfd,op,fd,&ee) == -1) return -1;
......
}
这里，我们就看到了epoll_ctl函数。
6、从epoll实例中获取就绪的事件
这个获取就绪事件的动作，是在main方法的aeMain函数中。
int main(int argc, char **argv) {
......
//执行aeMain函数开启事件循环处理框架
aeMain(server.el);
......
}
我们进入aeMain函数。
void aeMain(aeEventLoop *eventLoop) {
//只要redis实例没有停止，while循环就会一直执行
eventLoop->stop = 0;
//redis服务是否停止的标志,如果stop值变为1,说明redis服务停止了
while (!eventLoop->stop) {
......
aeProcessEvents(eventLoop, AE_ALL_EVENTS|AE_CALL_AFTER_SLEEP);
}
}
我们看到获取就绪的事件函数是aeProcessEvents，我们进入其中看一下
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
......
//调用多路复用API
numevents = aeApiPoll(eventLoop, tvp);
......
}
可以看到一个aeApiPoll函数，该函数是对epoll_wait函数的封装，我们继续进入aeApiPoll函数。
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;
//EPOLLIN代表epoll模型的读事件,这一行代码的意思是将epoll的读事件映射到redis事件驱动框架的读事件
if (e->events & EPOLLIN) mask |= AE_READABLE;
//EPOLLOUT代表epoll模型的写事件,这一行代码的意思是将epoll的写事件映射到redis事件驱动框架的写事件
if (e->events & EPOLLOUT) mask |= AE_WRITABLE;
//EPOLLERR:错误事件,表示文件描述符对应套接字出错
if (e->events & EPOLLERR) mask |= AE_WRITABLE;
if (e->events & EPOLLHUP) mask |= AE_WRITABLE;
//将epoll模型中已就绪的描述符映射到redis事件循环框架的就绪事件数组中
eventLoop->fired[j].fd = e->data.fd;
//给已就绪的事件设置事件类型
eventLoop->fired[j].mask = mask;
}
}
return numevents;
}
在其中，我们看到了epoll_wait函数，epoll_wait一共四个入参。
第一个是：要监听的描述符集合，第二个是要监听的事件类型，第三个是要监听的描述符数量，第四个是等待结果返回的超时时间
返回了结果后，下面的逻辑就是处理这个就绪的事件，这个方法是redis IO多路复用的关键所在，redis不停的接收客户端请求，这个方法是主要逻辑，我给每一行代码都加了注释，可以细看一下。
redis的这部分多路复用逻辑写的很清晰，可以认真梳理一下，对多路复用的原理会有更清晰的认识。
文章参考了极客时间的redis源码课程《redis源码剖析与实战》，文章写的挺好，有兴趣的小伙伴可以去看看。
