Redis所实现的Reactor模型设计方案

更新时间：2024年06月26日 14:55:14 作者：shark-chili

这篇文章主要介绍了Redis所实现的Reactor模型,本文将带领读者从源码的角度来查看redis关于reactor模型的设计,需要的朋友可以参考下

写在文章开头

我们都知道解决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)
{
	 //......
	 //非阻塞获取就绪事件
        numevents = 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处理新连接，如果是客户端则指向readQueryFromClient处理客户端的命令。
wfileProc:该指针服务端和客户端都一样，指向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模型内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Redis整合Spring结合使用缓存实例
这篇文章主要介绍了Redis整合Spring结合使用缓存实例,介绍了如何在Spring中配置redis，并通过Spring中AOP的思想，将缓存的方法切入到有需要进入缓存的类或方法前面。需要的朋友可以参考下
2015-12-12
Redis动态字符串SDS的实现
SDS在Redis中是实现字符串对象的工具,本文主要介绍了Redis动态字符串SDS的实现,具有一定的参考价值,感兴趣的可以了解一下
2023-11-11
Redis连接池监控（连接池是否已满）与优化方法
本文详细讲解了如何在Linux系统中监控Redis连接池的使用情况,以及如何通过连接池参数配置、系统资源使用情况、Redis命令监控、外部监控工具等多种方法进行检测和优化,以确保系统在高并发场景下的性能和稳定性,讨论了连接池的概念、工作原理、参数配置,以及优化策略等内容
2024-09-09
Redis缓存-序列化对象存储乱码问题的解决
这篇文章主要介绍了Redis缓存-序列化对象存储乱码问题的解决方案，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
2021-06-06
基于Redis的分布式锁的简单实现方法
这篇文章主要介绍了基于Redis的分布式锁的简单实现方法，Redis官方给出两种思路，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2018-10-10
redis实现延迟任务的项目实践
本文主要介绍了redis实现延迟任务的项目实践,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
2023-07-07
基于Redis分布式BitMap的应用分析
这篇文章主要介绍了基于Redis分布式BitMap的应用,本文给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2022-03-03
Redis从单点到集群部署模式(单机模式主从模式哨兵模式)
这篇文章主要为大家介绍了Redis从单点集群部署模式(单机模式主从模式哨兵模式)详解,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
2023-11-11
Redis集群详解
这篇文章主要介绍了Redis集群详解,需要的朋友可以参考下
2020-07-07
Redis高并发防止秒杀超卖实战源码解决方案
本文主要介绍了Redis高并发防止秒杀超卖实战源码解决方案，文中通过示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2021-10-10