当前位置： 首页 > news >正文

io多路复用：reactor模型的封装及与上层简单业务的实现（webserver）

news 2025/10/2 6:59:03

reactor模型

reacter模型就是将之间的epoll的多路复用版本做了一个简单的封装：

1.  大体分两种网络io：一种是监听io（绑定在不同端口上的socket套接字，且由listen开通），一种是负责处理连接收发信息请求的io（accept获取的套接字），将这些io分装成一个统一的控制块儿：

struct conn {

    int fd;

    char rbuffer[BUFFER_LENGTH];

    int rlength;

    char wbuffer[BUFFER_LENGTH];

    int wlength;

    RCALLBACK send_callback;

    union {

        RCALLBACK recv_callback;

        RCALLBACK accept_callback;

    } r_action;

    int status;

    char *payload;

    char mask[4];

};

        每个控制块，定义两个行为回调函数。可读和可写触发时发生相应的回调。也就是又事件去驱动回调。

比如接收的回调：（可写触发）

int recv_cb(int fd){memset(conn_list[fd].rbuffer, 0, sizeof(BUFFER_LENGTH));int count = recv(fd, conn_list[fd].rbuffer, BUFFER_LENGTH, 0);// 处理错误情况比如客户端异常退出大量连接不能正常断开if (count < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) {return 0; // 暂时没有数据}printf("recv error: %s\n", strerror(errno));epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);close(fd);return -1;}// 处理正常断开if (count == 0) {printf("client disconnect: %d\n", fd);epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);close(fd);return 0;}// 安全处理接收到的数据conn_list[fd].rlength = count;conn_list[fd].rbuffer[count] = '\0'; // 确保字符串结尾// printf("RECV: %s, length: %d\n", conn_list[fd].rbuffer, count);// conn_list[fd].wlength = conn_list[fd].rlength;// memcpy(conn_list[fd].wbuffer, conn_list[fd].rbuffer, conn_list[fd].wlength);// set_event(fd, EPOLLOUT, 0);http_request(&conn_list[fd]);// ws_request(&conn_list[fd]);set_event(fd, EPOLLOUT, 0);return count;   
}

发送的回调：（可写触发）

int send_cb(int fd){http_response(&conn_list[fd]);//和业务隔离// ws_response(&conn_list[fd]);int count = 0;if(conn_list[fd].status == 1){count = send(fd, conn_list[fd].wbuffer, conn_list[fd].wlength, 0);set_event(fd, EPOLLOUT, 0);}else if(conn_list[fd].status == 2){set_event(fd, EPOLLOUT, 0);}else if(conn_list[fd].status == 0){// if (conn_list[fd].wlength != 0){//     // count = send(fd, conn_list[fd].wbuffer, conn_list[fd].wlength, 0);// }set_event(fd, EPOLLIN, 0);}return count;
}

处理连接请求的回调：

int accept_cb(int fd){struct sockaddr_in clientaddr;socklen_t len = sizeof(clientaddr);int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len);if(clientfd < 0) {printf("accpet errno: %d\n", errno);return -1;}if((clientfd % 1000) == 0){struct timeval current;gettimeofday(&current, NULL);int time_used = TIME_SUB_MS(current, begin);memcpy(&begin, &current, sizeof(struct timeval));printf("accept finished: %d, timeused %d\n", clientfd, time_used);}// printf("accept finished :%d\n", clientfd);event_register(clientfd, EPOLLIN);// event_register(clientfd, EPOLLIN | EPOLLET);return 0;
}

注意比如这里发送回调函数就只处理发送的io的处理只负责接法数据的行为，和发送了什么这种业务处理都隔离开了，有利于业务的可扩展性。

测试百万并发

具体操作见简单实现Tcp服务器的百万并发-CSDN博客

主要就是设置文件打开的最大数值：

ulimit -n 1048576

然后设置一下系统文件

sudo vim /etc/sysctl.conf 

添加设置内容：

net.ipv4.ip_local_port_range = 1024 65535
net.ipv4.tcp_mem = 524288 1048576 1572864
net.ipv4.tcp_wmem = 2048 2048 4096
net.ipv4.tcp_rmem = 2048 2048 4096
fs.file-max = 1048576
net.netfilter.nf_conntrack_max = 1048576
net.netfilter.nf_conntrack_tcp_timeout_established = 1200

生效一下

sudo sysctl -p

这里操作的时候遇到了服务端被杀死退出的情况

发现是内存申请太多了，我们改小下程序申请的内存，100万个conn结构里里面buffer length就1024，改成了100 连接才上去了

测吞吐量qps

吞吐量（QPS）是衡量 Web 服务器处理请求能力的核心指标，全称是 Queries Per Second（每秒查询数），简单说就是服务器每秒能成功响应的 HTTP 请求数量。

这里我们用wrk工具

git clone https://github.com/wg/wrk.git

cd wrk

make
sudo cp wrk /usr/local/bin

测试命令

./wrk -c50 -t10 -d10s http://192.168.150.137:2000/

各参数解析：

  1. -c50

    • c 是 connections 的缩写，表示 并发连接数。
    • 这里设置为 50，意味着测试期间会持续保持与服务器的 50 个并发 TCP 连接。

  2. -t10

    • t 是 threads 的缩写，表示 测试使用的线程数。
    • 这里设置为 10，即启动 10 个工作线程来处理这 50 个并发连接（线程数通常建议设置为与 CPU 核心数相近，过多可能导致线程切换开销）。

  3. -d10s

    • d 是 duration 的缩写，表示 测试持续时间。
    • 这里设置为 10s，即测试会持续运行 10 秒。

  4. http://192.168.150.137:2000/

    • 目标测试的 URL，即向 IP 为 192.168.150.137、端口为 2000 的 Web 服务器发送 HTTP 请求，路径为根目录 /。

只回复一小段儿数据

int http_response(struct conn *c) {// 使用sprintf将HTTP响应内容格式化为字符串，存入c->wbuffer，并记录响应长度到c->wlengthc->wlength = sprintf(c->wbuffer,"HTTP/1.1 200 OK\r\n"  // HTTP协议版本和成功状态码"Content-Type: text/html\r\n"  // 响应内容类型为HTML"Accept-Ranges: bytes\r\n"  // 支持字节范围请求"Content-Length: 82\r\n"  // 响应体的字节长度"Date: Tue, 30 Apr 2024 13:16:46 GMT\r\n\r\n"  // 响应生成时间"<html><head><title>0voice.king</title></head><body><h1>King</h1></body></html>\r\n\r\n");return c->wlength;
}

测试得到：

去掉服务端缓冲区的打印信息：

发现qps上升了很多。可见打印信息还是很占时间的。

        此处我们发现会发现每次wrk测试完退出后，服务端也直接崩溃退出了。

发现问题所在：

        客户端直接关闭是会回发一个reset也就是会让服务端的recv返回-1，reset也是tcp八个状态之一，注意这是阻塞io的情况。非阻塞io返回-1还有EAGAIN和EWOULDBLOCK的现象。

水平触发和边缘触发

测试代码：


#include <errno.h>      // 定义errno
#include <netinet/in.h> // 网络地址结构
#include <poll.h>
#include <pthread.h>
#include <stdio.h>  // printf函数
#include <string.h> // strerror函数（关键补充）不加能运行但是输出strerror会段错误
#include <sys/epoll.h>
#include <sys/select.h>
#include <sys/socket.h> // socket相关函数
#include <sys/time.h>   // 新增：用于gettimeofday函数
#include <unistd.h>     // close函数
typedef int (*RCALLBACK)(int fd);#define BUFFER_LENGTH 10
#define CONNECTION_SIZE 1000000
#define MAXPORT 10
#define TIME_SUB_MS(tv1, tv2)  ((tv1.tv_sec - tv2.tv_sec) * 1000 + (tv1.tv_usec - tv2.tv_usec) / 1000)
int epfd;struct timeval begin;struct conn {int fd;char rbuffer[BUFFER_LENGTH];int rlength;char wbuffer[BUFFER_LENGTH];int wlength;RCALLBACK send_callback;union {RCALLBACK recv_callback;RCALLBACK accept_callback;} r_action;
};struct conn conn_list[CONNECTION_SIZE] = {0};int init_server(unsigned short port){int sockfd = socket(AF_INET, SOCK_STREAM, 0);if (sockfd == -1) {printf("socket create failed: %s\n", strerror(errno));return 1;}struct sockaddr_in servaddr;servaddr.sin_family = AF_INET;servaddr.sin_addr.s_addr = htonl(INADDR_ANY);servaddr.sin_port = htons(port);if (-1 == bind(sockfd, (struct sockaddr*)&servaddr, sizeof(servaddr))) {printf("bind failed: %s\n", strerror(errno));close(sockfd);return 1;}if (-1 == listen(sockfd, 10)) {printf("listen failed: %s\n", strerror(errno));close(sockfd);return 1;}printf("listen finished: %d\n", sockfd);return sockfd;
}int set_event(int fd, int event, int flag){if (flag) { //addstruct epoll_event ev;ev.data.fd = fd;ev.events = event;epoll_ctl(epfd, EPOLL_CTL_ADD, fd, &ev);}else{struct epoll_event ev;ev.events = event;ev.data.fd = fd;epoll_ctl(epfd, EPOLL_CTL_MOD, fd, &ev);}
}int accept_cb(int fd){struct sockaddr_in clientaddr;socklen_t len = sizeof(clientaddr);int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len);if(clientfd < 0) {printf("accpet errno: %d\n", errno);return -1;}if((clientfd % 1000) == 0){struct timeval current;gettimeofday(&current, NULL);int time_used = TIME_SUB_MS(current, begin);memcpy(&begin, &current, sizeof(struct timeval));printf("accept finished: %d, timeused %d\n", clientfd, time_used);}// printf("accept finished :%d\n", clientfd);// event_register(clientfd, EPOLLIN);event_register(clientfd, EPOLLIN | EPOLLET);return 0;
}int recv_cb(int fd){memset(conn_list[fd].rbuffer, 0, sizeof(BUFFER_LENGTH));int count = recv(fd, conn_list[fd].rbuffer, BUFFER_LENGTH, 0);// 处理错误情况比如客户端异常退出大量连接不能正常断开if (count < 0) {if (errno == EAGAIN || errno == EWOULDBLOCK) {return 0; // 暂时没有数据}printf("recv error: %s\n", strerror(errno));epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);close(fd);return -1;}// 处理正常断开if (count == 0) {printf("client disconnect: %d\n", fd);epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL);close(fd);return 0;}// 安全处理接收到的数据conn_list[fd].rlength = count;conn_list[fd].rbuffer[count] = '\0'; // 确保字符串结尾printf("RECV: %s, length: %d\n", conn_list[fd].rbuffer, count);// conn_list[fd].wlength = conn_list[fd].rlength;// memcpy(conn_list[fd].wbuffer, conn_list[fd].rbuffer, conn_list[fd].wlength);// set_event(fd, EPOLLOUT, 0);return count;   
}int send_cb(int fd){int count = send(fd, conn_list[fd].wbuffer, conn_list[fd].wlength, 0);set_event(fd, EPOLLIN, 0);return count;
}int event_register(int fd, int event){if(fd < 0 || fd > CONNECTION_SIZE) return -1;conn_list[fd].fd = fd;conn_list[fd].r_action.recv_callback = recv_cb;conn_list[fd].send_callback = send_cb;memset(conn_list[fd].rbuffer, 0, BUFFER_LENGTH);conn_list[fd].rlength = 0;memset(conn_list[fd].wbuffer, 0, BUFFER_LENGTH);conn_list[fd].wlength = 0;set_event(fd, event, 1);return 0;
}int main(int argc, char **argv) {if (argc <= 1) {printf("Usage: %s ip port\n", argv[0]);exit(0);}unsigned short port = atoi(argv[1]);// unsigned short port = 8000;epfd = epoll_create(1);int i = 0;for (i = 0; i < MAXPORT; i++) {int sockfd = init_server(port + i);conn_list[sockfd].fd = sockfd;conn_list[sockfd].r_action.recv_callback = accept_cb;set_event(sockfd, EPOLLIN, 1);}gettimeofday(&begin, NULL);while(1) {struct epoll_event events[1024] = {0};int nready = epoll_wait(epfd, events, 1024, -1);int i = 0;for(i = 0; i < nready; i++){int connfd = events[i].data.fd;if(events[i].events & EPOLLIN){conn_list[connfd].r_action.recv_callback(connfd);}if(events[i].events & EPOLLOUT){conn_list[connfd].send_callback(connfd);}}}return 0;
}

还是reactor这套代码，我们把BUFFER_LENGTH改成10（一次只能接收10个字符） event_register(clientfd, EPOLLIN | EPOLLET); 改成边缘触发发现：

        每次点击发送都只能接收10个字符就不动不接收了，甚至余下的数据都不断积压在后面，可以发现下一次的循环字符串的开头跟上一串的结尾都连一起发送了，说明每次发送的数据都直接存在了服务端接受缓冲区的里面等着用户层调用recv去取，取多少就释放多少，不取就一直积压在接收缓冲区。说明每次来新的数据才回触发而且只触发一次。

        再切换成之前的水平触发。

        我们发现每次发送，recv会直接调用连续调用三次知道吧数据接收完，说明了水平触发只要是接收缓冲区有数据就一直触发io，直到没数据为止。

        所以这里我们边缘触发采用while循环去接受数据，并且我们然后接受数据的io调成非阻塞的，因为再循环中阻塞会让进程挂起，同样非阻塞io也更适合边缘触发。

        listen也是典型的水平触发，只要有连接请求就不停地触发回调让accept去取相应的连接套接字。

适合场景：

边缘触发：适合包的大小不确定

水平触发：适合包的大小确定的

简单webserver业务的添加

http_request（处理接收请求，比如接收打印浏览器发来的请求）

比如用浏览器访问

终端返回的打印信息：

http_response处理回发浏览器的内容（比如一个html）

这里如果回发的很少一次就能回发完（小于发送缓冲区的极限）那就直接写在wbuffer里面，然后注册可读事件，等待回调函数发送

int http_response(struct conn *c) {// 使用sprintf将HTTP响应内容格式化为字符串，存入c->wbuffer，并记录响应长度到c->wlengthc->wlength = sprintf(c->wbuffer,"HTTP/1.1 200 OK\r\n"  // HTTP协议版本和成功状态码"Content-Type: text/html\r\n"  // 响应内容类型为HTML"Accept-Ranges: bytes\r\n"  // 支持字节范围请求"Content-Length: 82\r\n"  // 响应体的字节长度"Date: Tue, 30 Apr 2024 13:16:46 GMT\r\n\r\n"  // 响应生成时间"<html><head><title>0voice.king</title></head><body><h1>King</h1></body></html>\r\n\r\n");return c->wlength;
}

如过时大的文件就用sendfile（跳过用户层的wbuffer把数据，内核态内部的一次拷贝）

sendfile只负责发送请求体的内容

所以还有请求头的部分要处理所以这里设计了一个状态机：status，主要用来协调请求头和请求体的发送，主要用在http_response和send_cb里面

注册可读事件后，可读事件触发调用send_cb:

        status=0：在http_response里面构建请求头，存在wbuffer里面。回到send_cb后调用send把wbuffer里的数据传到发送缓冲区。然后status置1。

        status=1:   http_response用sendfile把文件中的数据作为请求体直接传输到发送缓冲区。status置2。

        status=2：清空wbuffer注册可读事件结束发送。

int http_response(struct conn *c) {int filefd = open("index.html", O_RDONLY);struct stat stat_buf;fstat(filefd, &stat_buf);// 0是开始 1是正在持续 2是结束if (c->status == 0) {c->wlength = sprintf(c->wbuffer,"HTTP/1.1 200 OK\r\n"         // HTTP协议版本和成功状态码"Content-Type: text/html\r\n" // 响应内容类型为HTML"Accept-Ranges: bytes\r\n"    // 支持字节范围请求"Content-Length: %ld\r\n"     // 响应体的字节长度"Date: Tue, 30 Apr 2024 13:16:46 GMT\r\n\r\n", // 响应生成时间stat_buf.st_size);c->status = 1;} else if (c->status == 1) {/*在两个文件描述符之间直接传输数据，无需通过用户空间缓冲区，属于“零拷贝”（zero-copy） 操作的一种，能显著提升数据传输效率。*/int ret = sendfile(c->fd, filefd, NULL, stat_buf.st_size);if (ret == -1) {printf("sendfile errr %d\n", errno);}printf("ret: %d stat_buf.st_size: %ld\n", ret, stat_buf.st_size);// c->wlength = 0;// memset(c->wbuffer, 0, BUFFER_LENGTH);c->status = 2;} else if (c->status == 2) {c->wlength = 0;memset(c->wbuffer, 0, BUFFER_LENGTH);c->status = 0;}/*BUFFER_LENGTH - c->wlength限制 read 最多往 buf 里写多少字节如果缓冲区大小（count）足够大（大于等于文件剩余未读数据量），单次 read 调用会直接从c->wbuffer +c->wlength读到文件末尾，并返回实际读取的字节数（等于剩余数据量）。*/close(filefd);return c->wlength;
}

        返回的是图片就改下请求头的

 "Content-Type: image/jpeg\r\n" // 响应内容类型为HTML

就可以了。

http://www.dtcms.com/a/431013.html

相关文章：

  • Pod 的 init Containers
  • 个人网站备案 导航单位门户网站怎么做
  • 网站建设充值入口pc营销型网站
  • ML4T - 第7章第6节 使用Alphalens进行分析 Alphalens Analysis
  • Lazarus下载和安装教程（附安装包，图文并茂）
  • 高端自适应网站设计深圳市年年卡网络科技公司是什么
  • 安卓基础组件017--ImageView组件
  • wordpress 设置多域名 一个站点wordpress个人主页主题
  • 少儿舞蹈小程序（21）我的页面搭建
  • 小程序前端功能更新说明
  • Cartograph＋explore_lite未知地形建图
  • FileLocator Pro(文件搜索工具) 多语便携版
  • 兼职做任务的网站免费网站下载直播软件免费
  • RabbitMQ死信交换机：消息的“流放之地“
  • LeetCode每日一题——加1
  • BriLLM框架研究可行性分析
  • 苏州工程网站建设wordpress导航菜单最右边
  • Java SE “泛型 + 注解 + 反射”面试清单（含超通俗生活案例与深度理解）
  • 22408计算机网络（初学）
  • 关于docker pull不了相关资源
  • OSPF Authentication-mode 概念
  • 网站怎么搭建在线编程网站开发
  • 以江协科技STM32入门教程的方式打开FreeRTOS——STM32C8T6如何移植FreeRTOS
  • 企业建设网站有哪些费用网站设计培训学院
  • ORB_SLAM2原理及代码解析：Frame::UnprojectStereo() 函数
  • SLAM算法分类对比
  • 碎片笔记｜生成模型原理解读：AutoEncoder、GAN 与扩散模型图像生成机制
  • 中文粤语（广州）语音语料库：6219条高质量语音数据助力粤语语音识别与自然语言处理研究
  • Kubernetes HTTPS迁移：Ingress到GatewayAPI实战
  • [Power BI] 矩阵表