目录
- 写在文章开头
- 详解Redis中的Reactor模型
- Reactor模型扫盲
- 详解reactor的实现
- 详解事件的封装
- 轮询并分发到handler
- 小结
写在文章开头
我们都知道解决C10k问题的最好方案就是通过在IO多路复用的基础上通过reactor模型实现高性能的网络并发程序，借助这个设计，redis的主线程也是基于IO多路复用以reactor模型的思路实现了一个高性能的单线程内存数据，本文将带领读者从源码的角度来查看redis关于reactor模型的设计。
详解Redis中的Reactor模型
Reactor模型扫盲
在此之前我们先来了解一下Reactor模型，在高性能网络并发程序的设计中，Reactor模型通过reactor接收用户连接事件、读事件、写事件这些网络事件，得到连接事件之后通过acceptor为其分配handler，后续的这些客户端的读写事件都会交由handler完成读写事件的处理，由此实现尽可能少的线程处理尽可能多的连接。
详解reactor的实现
上文我们简单的对Reactor模型进行了简单的扫盲，接下来我们将从redis的源码来了解redis对于Reactor模型的实现，我们都知道Reactor模型是通过reactor接收连接、读、写三种事件的，这一点我们可以直接在main方法看到aeMain的调用，该方法内部本质就是通过epoll模型进行非阻塞获取就的网络事件：
int main(int argc, char **argv) {
//前置初始化步骤
//......
//事件循环轮询前置操作
aeSetBeforeSleepProc(server.el,beforeSleep);
//执行事件驱动框架，循环处理各种触发的事件
aeMain(server.el);
//事件循环后置操作
aeDeleteEventLoop(server.el);
return 0;
}
我们步入aeMain方法，可以看到只要eventLoop没有停止就会无限循环调用aeProcessEvents获取并处理就绪的事件：
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
while (!eventLoop->stop) {
//......
//轮询并处理就绪的事件
aeProcessEvents(eventLoop, AE_ALL_EVENTS);
}
}
步入aeProcessEvents方法，我们就可以看到redis通过对于epoll的封装函数aeApiPoll非阻塞获取就绪的IO事件，注意笔者所强调的非阻塞获取，这也就是为什么redis仅仅用一个主线程即可实现Reactor模型的原因所在。
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
//......
//非阻塞获取就绪事件
nu编程客栈mevents = aeApiPoll(eventLoop, tvp);
for (j = 0; j < numevents; j++) {
//......
//处理事件
processed++;
}
}
/* Check time events */
if (flags & AE_TIME_EVENTS)
processed += processTimeEvents(eventLoop);
return processed; /* return the number of processed file/time events */
}
对此我们再次步入aeApiPoll实现可以看到redis对于epoll的调用epoll_wait，得到事件数retval 之后，直接基于retval遍历eventLoop的events这里面存储的就是所有收到的事件aeFiredEvent，redis会根据其事件类型累加对应的事件mask值，例如如果是得到的事件类型是EPOLLIN则mask值会加上AE_READABLE(1)，若是标准输出事件EPOLLOUT则累加AE_WRITABLE即2:
对应的我们给出这段基于epoll实现reacor的实现，可以看到其reactor通过事件轮询获取对应的事件类型再将其封装为aeFileEvent存到事件数组eventLoop->fired中：
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;
//根据事件类型累加读写的mask值
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;
//将该事件存到fired数组中
eventLoop->fired[j].fd = e->data.fd;
eventLoop->fired[j].mask = mask;
}
}
//返回事件数
return numevents;
}
详解事件的封装
上文我们提到一个aeFileEvent 事件的概念，该个事件结构如下图所示，它通过mask标记当前IO事件类型，在epoll轮询到事件时，它并通过rfileProc读事件处理指针和wfileProc写文件处理保存针对网络IO事件的处理函数，注意这个处理函数我们完全可以直接理解为reactor模型中的handler，最后用clientData记录客户端私有数据的指针：
typedef struct aeFileEvent {
//记录事件读写类型，如果是读事件READABLE则mask+1，若是写事件WRITABLE则加2
int mask; /* one of AE_(READABLE|WRITABLE) */
//读事件处理器指针指向读事件处理函数handler
aeFileProc *rfileProc;
//写事件处理器指针指向读事件处理函数handler
aeFileProc *wfileProc;
//记录客户端私有数据指针
编程客栈 void *clientData;
} aeFileEvent;
这里我们以服务端socket初始化阶段为例展示一下aeFileEvent对应处理器的初始化过程，我们在redis服务端启动的main函数可以看到initServer的调用，该方法会为当前服务端socket套接字的文件描述符绑定读事件的处理器acceptTcpHandler：
对应的我们给出这一段事件绑定handler的逻辑的核心代码段:
int main(int argc, char **argv) {
//......
//server初始化，其内部会完成数据结构、键值对数据库初始化、网络框架初始化工作
initServer();
}
void initServer(void) {
//......
for (j = 0; j < server.ipfd_count; j++) {
//为每一个监听服务端socket的读事件绑定对应的TCP处理器acceptTcpHandler，并将其注册到eventLoop中
if (aeCreateFileEvent(server.el, server.ipfd[j], AE_READABLE,
acceptTcpHandler,NULL) == AE_ERR)
{
redisPanic(
"Unrecoverable error creating server.ipfd file event.");
}
}
//......
}
轮询并分发到handler
上述步骤完成redis server的事件注册之后，main方法的aeMain函数就会通过epoll轮询eventLoop中是否有就绪的IO事件，如果redis server的fd的读事件就绪就会交给当前对应的读处理器完成redis客户端初始化工作，后续redis客户端套接字的fd也会将读写事件注册到eventLoop中，如此一来所有的服务端和客户端socket的读写事件都会注册到epoll上，让epoll作为reactor进行轮询，然后根据读写事件分配到各自的handler即rfileProc/wfileProc 指针所指向的函数上。
rfileProc/wfileProc指针指向的函数列表:
rfileProc:如果是redis服务端则该指针指向acceptTcpHandler处理新http://www.devze.com连接，如果是客户端则指向readQueryFromClient处理客户端的命令。wfiwww.devze.comleProc:该指针服务端和客户端都一样，指向sendReplyToClient用于将响应结果发送给客户端。
对应的我们给出上述描述的核心代码段，可以看到main方法会调用aeMain开始事件轮询：
int main(int argc, char **argv) {
//前置初始化步骤
//......
//事件循环轮询前置操作
aeSetBeforeSleepProc(server.el,beforeSleep);
//执行事件驱动框架，循环处理各种触发的事件
aeMain(server.el);
//事件循环后置操作
aeDeleteEventLoop(server.el);
return 0;
}
步入aeMain即可看到无限循环传入eventLoop查看是否有就绪的事件：
void aeMain(aeEventLoop *eventLoop) {
eventLoop->stop = 0;
while (!eventLoop->stop) {
//......
//传入eventLoop查看是否有socket的事件就绪
aeProcessEvents(eventLoop, AE_ALL_EVENTS);
}
}
继续步入aeProcessEvents即看到轮询就绪事件、acceptor调用acceptTcpHandler分发到读写的处理器handler上、后续客户端都会基于读写handler完成事件处理这样一套核心的reactor模型设计：
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
{
//......
//调用epoll获取所有就绪的socket的读写事件
numevents = aeApiPoll(eventLoop, tvp);
for (j = 0; j < numevents; j++) {
//获取当前事件的读写类型为mask赋值
aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
int mask = eventLoop->fired[j].mask;
int fd = eventLoop->fired[j].fd;
int rfired = 0;
//如果是读事件则交给rfileProc指向的函数，可以是服务端socket的连接处理器acceptTcpHandler，也可能是客户端的命令处理器readQueryFromClient
if (fe->mask & mask & AE_READABLE) {
rfired = 1;
fe->rfileProc(eventLoop,fd,fe->clientData,mask);
}
//如果编程客栈是写事件则调用wfileProc指向的sendReplyToClient将结果发送给客户端
if (fe->mask & mask & AE_WRITABLE) {
if (!rfired || fe->wfileProc != fe->rfileProc)
fe->wfileProc(eventLoop,fd,fe->clientData,mask);
}
processed++;
}
}
//......
}
小结
自此我们将redis单线程的reactor模型设计都分析完成了，希望对你有帮助。
到此这篇关于Redis所实现的Reactor模型的文章就介绍到这了,更多相关Redis Reactor模型内容请搜索编程客栈(www.devze.com)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程客栈(www.devze.com)！
加载中,请稍侯......
精彩评论