深入理解Reactor模型的原理与应用

1、什么是Reactor模型

        Reactor意思是“反应堆”,是一种事件驱动机制。

        和普通函数调用的不同之处在于:应用程序不是主动的调用某个 API 完成处理,而是恰恰相反,Reactor逆置了事件处理流程,应用程序需要提供相应的接口并注册到 Reactor 上,如果相应的时间发生,Reactor将主动调用应用程序注册的接口,这些接口又称为“回调函数”。

        对于刚开始接触这个机制,个人感觉翻译成“感应器”可能会更好理解一点,因为注册在Reactor上的函数就像感应器一样,只要有事件到达,就会触发它开始工作。

        Reactor 模式是编写高性能网络服务器的必备技术之一。


2、Reactor模型的优点

  • 响应快,不必为单个同步时间所阻塞,虽然 Reactor 本身依然是同步的;
  • 编程相对简单,可以最大程度的避免复杂的多线程及同步问题,并且避免了多线程/进程的切换开销;
  • 可扩展性强,可以方便的通过增加 Reactor 实例个数来充分利用 CPU 资源;
  • 可复用性高,reactor 框架本身与具体事件处理逻辑无关,具有很高的复用性;
           Reactor 模型开发效率上比起直接使用 IO 复用要高,它通常是单线程的,设计目标是希望单线程使用一颗 CPU 的全部资源。
            优点即每个事件处理中很多时候可以不考虑共享资源的互斥访问。可是缺点也是明显的,现在的硬件发展,已经不再遵循摩尔定律,CPU 的频率受制于材料的限制不再有大的提升,而改为是从核数的增加上提升能力,当程序需要使用多核资源时,Reactor 模型就会悲剧 , 为什么呢?
            如果程序业务很简单,例如只是简单的访问一些提供了并发访问的服务,就可以直接开启多个反应堆,每个反应堆对应一颗 CPU 核心,这些反应堆上跑的请求互不相关,这是完全可以利用多核的。例如 Nginx 这样的 http 静态服务器。

3、通过对网络编程(epoll)代码的优化,深入理解Reactor模型

1、epoll的普通版本,根据fd类型(listen_fd和client_fd)分为两大类处理。

        如果是listen_fd,调用accept处理连接请求;

        如果是client_fd,调用recv或者send处理数据。

         代码实现:


#include <stdio.h>
#include <string.h>
#include <stdlib.h>#include <unistd.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <sys/epoll.h>#include <errno.h>int main(int argc, char* argv[])
{if (argc < 2)return -1;int port = atoi(argv[1]);   //字符串转换为整型int sockfd = socket(AF_INET, SOCK_STREAM, 0);if (sockfd < 0)return -1;struct sockaddr_in addr;memset(&addr, 0, sizeof(struct sockaddr_in));   //新申请的空间一定要置零addr.sin_family = AF_INET;addr.sin_port = htons(port);    //转换成网络字节序addr.sin_addr.s_addr = INADDR_ANY;if (bind(sockfd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in)) < 0)return -2;if (listen(sockfd, 5) < 0)return -3;//epollint epfd = epoll_create(1); //创建epoll,相当于红黑树的根节点struct epoll_event ev, events[1024] = {0};  //events相当于就绪队列,一次性可以处理的集合ev.events = EPOLLIN;ev.data.fd = sockfd;epoll_ctl(epfd, EPOLL_CTL_ADD, sockfd, &ev);    //将ev节点加入到epoll,此处的sockfd参数随便添加没有意义,需要操作系统索引和它有对应的句柄while (1){int nready = epoll_wait(epfd, events, 1024, -1);    //第四个参数-1表示一直等待,有事件才返回if (nready < 1) //没有事件触发,nready代表触发事件的个数break;int i = 0;for (i = 0; i < nready; i++)    //epoll_wait带出的就绪fd包括两大类:1、处理连接的listen_fd,2、处理数据的send和recv{if (events[i].data.fd == sockfd) //如果是listenfd,就将它加入到epoll{struct sockaddr_in client_addr;memset(&client_addr, 0, sizeof(struct sockaddr_in));socklen_t client_len = sizeof(client_addr);int client_fd = accept(sockfd, (struct sockaddr *)&client_addr, &client_len);if (client_fd <= 0)continue;char str[INET_ADDRSTRLEN] = {0};printf("recv from IP = %s ,at Port= %d\n", inet_ntop(AF_INET, &client_addr.sin_addr, str, sizeof(str)), ntohs(client_addr.sin_port));ev.events = EPOLLIN | EPOLLET;  //epoll默认是LT模式ev.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &ev);}else    //fd进行读写操作{//对fd的读写操作没有分开int client_fd = events[i].data.fd;char buf[1024] = {0};int ret = recv(client_fd, buf, 1024, 0);if (ret < 0){if (errno == EAGAIN || errno == EWOULDBLOCK){//}else{//}printf("ret < 0,断开连接:%d\n", client_fd);close(client_fd);ev.events = EPOLLIN;ev.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &ev);}else if (ret == 0)  //接收到了客户端发来的断开连接请求FIN后,没有及时调用close函数,进入了CLOSE _WAIT状态{printf("ret = 0,断开连接:%d\n", client_fd);close(client_fd);ev.events = EPOLLIN;ev.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &ev); //close关闭连接后要将它既是从epoll中删除}else{printf("Recv: %s, %d Bytes\n", buf, ret);}//区分fd的读写操作,即recv和sendif (events[i].events & EPOLLIN){int client_fd = events[i].data.fd;char buf[1024] = {0};int ret = recv(client_fd, buf, 1024, 0);if (ret < 0){if (errno == EAGAIN || errno == EWOULDBLOCK){//...}else{//...}printf("ret < 0,断开连接:%d\n", client_fd);close(client_fd);ev.events = EPOLLIN;ev.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &ev);}else if (ret == 0)  //接收到了客户端发来的断开连接请求FIN后,没有及时调用close函数,进入了CLOSE _WAIT状态{printf("ret = 0,断开连接:%d\n", client_fd);close(client_fd);ev.events = EPOLLIN;ev.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &ev); //close关闭连接后要将它既是从epoll中删除}else{printf("Recv: %s, %d Bytes\n", buf, ret);}}if (events[i].events & EPOLLOUT)    //为什么需要判断EPOLLOUT,而不是直接else?因为一个fd有可能同时存在可读和可写事件的{int client_fd = events[i].data.fd;char buf[1024] = {0};send(client_fd, buf, sizeof(buf), 0);}}}}return 0;
}

 

2、epoll的优化版本,根据事件类型(读和写)分为两大类处理。

         代码实现:

        for (i = 0; i < nready; i++)    //epoll_wait带出的就绪fd包括两大类:1、处理连接的listen_fd,2、处理数据的send和recv{//区分fd的读写操作if (events[i].events & EPOLLIN){if (events[i].data.fd == sockfd) //如果是listenfd,就将它加入到epoll{struct sockaddr_in client_addr;memset(&client_addr, 0, sizeof(struct sockaddr_in));socklen_t client_len = sizeof(client_addr);int client_fd = accept(sockfd, (struct sockaddr *)&client_addr, &client_len);if (client_fd <= 0)continue;char str[INET_ADDRSTRLEN] = {0};printf("recv from IP = %s ,at Port= %d\n", inet_ntop(AF_INET, &client_addr.sin_addr, str, sizeof(str)), ntohs(client_addr.sin_port));ev.events = EPOLLIN | EPOLLET;  //epoll默认是LT模式ev.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_ADD, client_fd, &ev);}else {int client_fd = events[i].data.fd;char buf[1024] = {0};int ret = recv(client_fd, buf, 1024, 0);if (ret < 0){if (errno == EAGAIN || errno == EWOULDBLOCK){//...}else{//...}printf("ret < 0,断开连接:%d\n", client_fd);close(client_fd);ev.events = EPOLLIN;ev.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &ev);}else if (ret == 0)  //接收到了客户端发来的断开连接请求FIN后,没有及时调用close函数,进入了CLOSE _WAIT状态{printf("ret = 0,断开连接:%d\n", client_fd);close(client_fd);ev.events = EPOLLIN;ev.data.fd = client_fd;epoll_ctl(epfd, EPOLL_CTL_DEL, client_fd, &ev); //close关闭连接后要将它既是从epoll中删除}else{printf("Recv: %s, %d Bytes\n", buf, ret);}}}//为什么需要判断EPOLLOUT,而不是直接else?因为一个fd有可能同时存在可读和可写事件的if (events[i].events & EPOLLOUT)    {int client_fd = events[i].data.fd;char buf[1024] = {0};send(client_fd, buf, sizeof(buf), 0);}}

 

3、epoll的Reactor模式, epoll由以前的对网络io(fd)进行管理,转变成对events事件进行管理。

         代码实现:

#include <stdio.h>
#include <string.h>
#include <stdlib.h>#include <unistd.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <sys/epoll.h>#include <errno.h>//每个fd所对应的信息
struct sockitem
{int sockfd;int (*callback)(int fd, int events, void*arg);char sendbuf[1024];char recvbuf[1024];
};//每个epoll所对应的信息
struct epollitem
{int epfd;struct epoll_event events[1024];    //events相当于就绪队列,一次性可以处理的集合
};struct epollitem *eventloop = NULL;int recv_cb(int fd, int events, void*arg);
int send_cb(int fd, int events, void*arg);int accept_cb(int fd, int events, void*arg)
{printf("---accept_cb(int fd, int events, void*arg)---\n");struct sockaddr_in client_addr;memset(&client_addr, 0, sizeof(struct sockaddr_in));socklen_t client_len = sizeof(client_addr);int client_fd = accept(fd, (struct sockaddr *)&client_addr, &client_len);if (client_fd <= 0)return -1;char str[INET_ADDRSTRLEN] = {0};printf("recv from IP = %s ,at Port= %d\n", inet_ntop(AF_INET, &client_addr.sin_addr, str, sizeof(str)), ntohs(client_addr.sin_port));struct epoll_event ev;ev.events = EPOLLIN | EPOLLET;  //epoll默认是LT模式struct sockitem *si = (struct sockitem*)malloc(sizeof(struct sockitem));si->sockfd = client_fd;si->callback = recv_cb;ev.data.ptr = si;epoll_ctl(eventloop->epfd, EPOLL_CTL_ADD, client_fd, &ev);return client_fd;
}int recv_cb(int fd, int events, void*arg)
{printf("---recv_cb(int fd, int events, void*arg)---\n");struct epoll_event ev;struct sockitem *sit = (struct sockitem*)arg;int ret = recv(fd, sit->recvbuf, 1024, 0);if (ret < 0){if (errno == EAGAIN || errno == EWOULDBLOCK){//...}else{//...}printf("ret < 0,断开连接:%d\n", fd);ev.events = EPOLLIN;epoll_ctl(eventloop->epfd, EPOLL_CTL_DEL, fd, &ev);    //close关闭连接后要将它既是从epoll中删除close(fd);free(sit);  //连接关闭后释放内存}else if (ret == 0)  //接收到了客户端发来的断开连接请求FIN后,没有及时调用close函数,进入了CLOSE _WAIT状态{printf("ret = 0,断开连接:%d\n", fd);ev.events = EPOLLIN;epoll_ctl(eventloop->epfd, EPOLL_CTL_DEL, fd, &ev); close(fd);free(sit);}else{printf("Recv from recvbuf:  %s, %d Bytes\n", sit->recvbuf, ret);ev.events = EPOLLIN | EPOLLOUT;  //sit->sockfd = fd;sit->callback = send_cb;ev.data.ptr = sit;epoll_ctl(eventloop->epfd, EPOLL_CTL_MOD, fd, &ev);}return ret;
}int send_cb(int fd, int events, void*arg)
{struct epoll_event ev;struct sockitem *sit = (struct sockitem*)arg;strncpy(sit->sendbuf, sit->recvbuf, sizeof(sit->recvbuf) + 1);send(fd, sit->sendbuf, sizeof(sit->recvbuf) + 1, 0);ev.events = EPOLLIN | EPOLLET;  //sit->sockfd = fd;sit->callback = recv_cb;ev.data.ptr = sit;epoll_ctl(eventloop->epfd, EPOLL_CTL_MOD, fd, &ev);return fd;
}int main(int argc, char* argv[])
{if (argc < 2)return -1;int port = atoi(argv[1]);   //字符串转换为整型int sockfd = socket(AF_INET, SOCK_STREAM, 0);if (sockfd < 0)return -1;struct sockaddr_in addr;memset(&addr, 0, sizeof(struct sockaddr_in));   //新申请的空间一定要置零addr.sin_family = AF_INET;addr.sin_port = htons(port);    //转换成网络字节序addr.sin_addr.s_addr = INADDR_ANY;if (bind(sockfd, (struct sockaddr*)&addr, sizeof(struct sockaddr_in)) < 0)return -2;if (listen(sockfd, 5) < 0)return -3;//epolleventloop = (struct epollitem *)malloc(sizeof(struct epollitem));eventloop->epfd = epoll_create(1); //创建epoll,相当于红黑树的根节点struct epoll_event ev;ev.events = EPOLLIN | EPOLLET;struct sockitem *si = (struct sockitem*)malloc(sizeof(struct sockitem));si->sockfd = sockfd;si->callback = accept_cb;ev.data.ptr = si;   //将fd和对应的回调函数绑定一起带进epollepoll_ctl(eventloop->epfd, EPOLL_CTL_ADD, sockfd, &ev);    //将ev节点加入到epoll,此处的sockfd参数随便添加没有意义,需要操作系统索引和它有对应的句柄while (1){int nready = epoll_wait(eventloop->epfd, eventloop->events, 1024, -1);    //第四个参数-1表示一直等待,有事件才返回if (nready < 1) //没有事件触发,nready代表触发事件的个数break;int i = 0;for (i = 0; i < nready; i++){//区分fd的读写操作if (eventloop->events[i].events & EPOLLIN){struct sockitem *sit = (struct sockitem*)eventloop->events[i].data.ptr;sit->callback(sit->sockfd, eventloop->events[i].events, sit);    //不用区分listen_fd和recv_fd,相应的fd都会调用他们所对应的callback}//为什么需要判断EPOLLOUT,而不是直接else?因为一个fd有可能同时存在可读和可写事件的if (eventloop->events[i].events & EPOLLOUT)    {struct sockitem *sit = (struct sockitem*)eventloop->events[i].data.ptr;sit->callback(sit->sockfd, eventloop->events[i].events, sit);}}}return 0;
}

4、Reactor模型的应用 

        1、单线程模式的Reactor,参考libevent、redis;

        2、多线程模式的Reactor,参考memcached;

        3、多进程模式的Reactor,参考nginx。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/113764.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

非科班菜鸡算法学习记录 | 代码随想录算法训练营第51天||309.最佳买卖股票时机含冷冻期 714.买卖股票的最佳时机含手续费 股票总结

309.最佳买卖股票时机含冷冻期 309. Best Time to Buy and Sell Stock with Cooldown(英文力扣连接) 知识点&#xff1a;动规 状态&#xff1a;看思路ok 思路&#xff1a; 四个状态需要想&#xff0c;持有/不持有且过了冷却期/当天卖/正处于冷却期&#xff1b; 具体看注释…

如何用Python爬虫持续监控商品价格

目录 持续监控商品价格步骤 1. 选择合适的爬虫库&#xff1a; 2. 选择目标网站&#xff1a; 3. 编写爬虫代码&#xff1a; 4. 设定监控频率&#xff1a; 5. 存储和展示数据&#xff1a; 6. 设置报警机制&#xff1a; 7. 异常处理和稳定性考虑&#xff1a; 可能会遇到的…

李跳跳下载-《告别广告困扰,让李跳跳助力打造清爽浏览体验》

大家好&#xff0c;&#x1f44b;今天我想向大家介绍一款非常好用的应用程序——李跳跳 App &#x1f680;。 随着智能手机的普及&#xff0c;应用程序已经成为了我们日常生活中必不可少的一部分。但是&#xff0c;随之而来的是各种各样的广告&#xff0c;这些广告不仅浪费我们…

交换机端口安全实验

文章目录 一、实验的背景与目的二、实验拓扑三、实验需求四、实验解法1. PC配置IP地址部分2. 在SW1上开启802.1X身份验证3. 创建一个用户身份验证的用户。用户名为wangdaye&#xff0c;密码为1234564.创建一个端口隔离组&#xff0c;实现三台PC无法互相访问 摘要&#xff1a; 本…

vue中使用window.open打开assets文件夹下的pdf文件

需求&#xff1a;系统有个操作手册&#xff0c;点击会在浏览器新开个窗口并打开pdf文件。这个pdf文件存储在本地assets文件夹中。 文件结构&#xff1a; 注&#xff1a;直接使用window.open(文件路径)不能打开&#xff0c;需要在vue.config.js中配置所需文件 引入图中红框中的…

Ruoyi微服务启动流程

1、执行sql 执行sql ry-quarty.sql ry_2023706.sql 到ry-cloud 数据库 2、下载nacos 修改配置文件 修改连接地址 启动nacos 看到下面的配置文件即为成功 修改配置文件里面的数据库连接信息 3、修改nacos 为单机启动 4、启动项目即可 nacos自取 链接: https://pan.baidu…

云上办公系统项目

云上办公系统项目 1、云上办公系统1.1、介绍1.2、核心技术1.3、开发环境说明1.4、产品展示后台前台 1.5、 个人总结 2、后端环境搭建2.1、建库建表2.2、创建Maven项目pom文件guigu-oa-parentcommoncommon-utilservice-utilmodelservice-oa 配置数据源、服务器端口号application…

Python钢筋混凝土结构计算.pdf-已知弯矩确定混凝土梁截面尺寸

计算原理 确定混凝土梁截面的合理尺寸通常需要考虑弯矩、受力要求和约束条件等多个因素。以下是一种常见的计算公式&#xff0c;用于基于已知弯矩确定混凝土梁截面的合理尺寸&#xff1a; 请注意&#xff0c;以上公式仅提供了一种常见的计算方法&#xff0c;并且具体的规范和设…

脚本掌控,Linux上实现Spring Boot(JAR包)开机自启

&#x1f60a; 作者&#xff1a; 一恍过去 &#x1f496; 主页&#xff1a; https://blog.csdn.net/zhuocailing3390 &#x1f38a; 社区&#xff1a; Java技术栈交流 &#x1f389; 主题&#xff1a; 脚本掌控&#xff0c;Linux上实现Spring Boot&#xff08;JAR包&#x…

城市内涝积水监测预警系统 yolov8

城市内涝积水监测预警系统通过yolov8网络深度学习框架&#xff0c;算法一旦识别到道路出现积水&#xff0c;城市内涝积水监测预警系统会立即发出预警信号。并及时通知相关人员。YOLO检测速度非常快。标准版本的YOLO可以每秒处理 45 张图像&#xff1b;YOLO的极速版本每秒可以处…

聚观早报|OpenAI宣布推出企业版ChatGPT;苹果公司开设8家新店

【聚观365】8月30日消息 OpenAI宣布推出企业版ChatGPT 比亚迪上半年净利润109.5亿元 歌尔股份上半年净利润4.22亿元 一起教育科技Q2营收6925万元 苹果公司今年开设8家新店 OpenAI宣布推出企业版ChatGPT 据外媒报道&#xff0c;当地时间周一&#xff0c;美国人工智能研究…

k8s的交付与部署案例操作

一 k8s的概念 1.1 k8s k8s是一个轻量级的&#xff0c;用于管理容器化应用和服务的平台。通过k8s能够进行应用的自动化部署和扩容缩容。 1.2 k8s核心部分 1.prod: 最小的部署单元&#xff1b;一组容器的集合&#xff1b;共享网络&#xff1b;生命周期是短暂的&#xff1b; …

QT概括-Rainy

Qt 虽然经常被当做一个 GUI 库&#xff0c;用来开发图形界面应用程序&#xff0c;但这并不是 Qt 的全部&#xff1b;Qt 除了可以绘制漂亮的界面&#xff08;包括控件、布局、交互&#xff09;&#xff0c;还包含很多其它功能&#xff0c;比如多线程、访问数据库、图像处理、音频…

【JasperReports笔记06】JasperReport报表开发之常见的组件元素(Table、Subreport、Barcode等)

这篇文章&#xff0c;主要介绍JasperReport报表开发之常见的组件元素&#xff08;Table、Subreport、Barcode等&#xff09;。 目录 一、基础组件元素 1.1、StaticText 1.2、TextField 1.3、Image 1.4、Break分页 1.5、Rectangle矩形区域 1.6、Ellipse椭圆区域 1.7、Li…

Mybatis中 list.size() = 1 但显示 All elements are null

一、Bug展示 二、原因分析 2.1.情形一&#xff1a;Mybatis的XML中返回类型映射错误 <select id"selectByDesc" parameterType"com.task.bean.OrderInfo"resultType"com.task.bean.OrderInfo">select MER_ID,SETTLE_DATE,ICE_NAME,ORDER_S…

探索未来世界,解密区块链奥秘!

你是否曾好奇&#xff0c;区块链是如何影响着我们的生活与未来&#xff1f;想要轻松了解这个引领着技术革命的概念吗&#xff1f;那么这本令人着迷的新书《区块链导论》绝对值得你拥有&#xff01; 内容丰富多彩&#xff0c;让你轻松掌握&#xff1a; **1章&#xff1a;区块链…

文件传输协议

文章目录 一、FTP1. 定义2. 端口3. 数据传输方式主动方式被动方式 二、TFTP三、常用命令 首先可以看下思维导图&#xff0c;以便更好的理解接下来的内容。 一、FTP 1. 定义 文件传输协议&#xff08;FTP&#xff09;是一种用于在客户端和服务器之间进行文件传输的标准网络协…

使用yarn build 打包vue项目时静态文件或图片未打包成功

解决Vue项目使用yarn build打包时静态文件或图片未打包成功的问题 1. 检查vue.config.js文件 首先&#xff0c;我们需要检查项目根目录下的vue.config.js文件&#xff0c;该文件用于配置Vue项目的打包和构建选项。在这个文件中&#xff0c;我们需要确认是否正确地配置了打包输…

openCV实战-系列教程13:文档扫描OCR识别下(图像轮廓/模版匹配)项目实战、源码解读

&#x1f9e1;&#x1f49b;&#x1f49a;&#x1f499;&#x1f49c;OpenCV实战系列总目录 有任何问题欢迎在下面留言 本篇文章的代码运行界面均在Pycharm中进行 本篇文章配套的代码资源已经上传 上篇内容&#xff1a; openCV实战-系列教程11&#xff1a;文档扫描OCR识别上&am…

【Linux】0基础从获取docker,一步一步到部署PaddleSpeech

一、利用VMware安装ubuntu 1.安装VMware 具体操作详细安装VMware的方式 另外附部分VMware密匙 4A4RR-813DK-M81A9-4U35H-06KND NZ4RR-FTK5H-H81C1-Q30QH-1V2LA JU090-6039P-08409-8J0QH-2YR7F 4Y09U-AJK97-089Z0-A3054-83KLA 4C21U-2KK9Q-M8130-4V2QH-CF810 MC60H-DWH…