如何将epoll的IO驱动封装成reactor事件反应堆驱动
Reactor 模式是处理并发 I/O 比较常见的一种模式,用于同步 I/O,中心思想是将所有要处理的 I/O 事件注册到一个 I/O 多路复用器上,同时主线程/进程阻塞在多路复用器上; 一旦有 I/O 事件到来或是准备就绪(文件描述符或 socket 可读、写),多路复用器返回并将事先注册的相应 I/O 事件分发到对应的处理器中。 组件
流程
首先我们肯定需要传入的是 fd 作为参数, 然后我们需要传入事件类型events 还有我们需要传入sockitem 结构体指针, 因为如果是读IO事件我们需要将从客户端读取的数据写入到sockitem的处在用户空间的recvbuffer中去, 以及如果是写IO事件我们需要将sockitem的处在用户空间的sendbuffer中的数据写回客户端 然后针对返回值我们设置为int类型即可, 所以接口设计为了如下结果,
首先我们肯定需要一个epoll句柄,所以epfd肯定需要封装进去,其次我们需要一个容器存储触发的IO事件,至此我们应该设置一个 sruct epoll_event events[512];在其中存储触发的IO事件,也就是将所有需要的全局数据封装成reactor
accept_cb : 新连接到来事件处理器
recv_cb : 处理读事件的处理器
send_cb 写事件处理器
#include <stdio.h> #include <stdlib.h> #include <unistd.h> #include <sys/types.h> #include <sys/socket.h> #include <string.h> #include <arpa/inet.h> #include <sys/epoll.h> #include <errno.h> #include <fcntl.h> typedef struct sockaddr SA; #define BUFFSIZE 1024 struct sockitem { int sockfd; //事件处理器,处理函数回调接口 int (*callback)(int fd, int events, void* arg); //读写函数 char recvbuffer[BUFFSIZE]; char sendbuffer[BUFFSIZE]; //读写字节数 int rlen; int slen; }; struct reactor { int epfd; struct epoll_event events[512]; }; //定义全局的eventloop --> 事件循环 struct reactor* eventloop = NULL; //申明这些事件处理器函数 int recv_cb(int fd, int events, void *arg); int accept_cb(int fd, int events, void* arg); int send_cb(int fd, int evnts, void* arg); int recv_cb(int fd, int events, void *arg) { struct sockitem* si = (struct sockitem*)arg; struct epoll_event ev;//后面需要 //处理IO读事件 int ret = recv(fd, si->recvbuffer, BUFFSIZE, 0); if (ret < 0) { if (errno == EAGAIN || errno == EWOULDBLOCK) { // return -1; } else { } //出错了,从监视IO事件红黑树中移除结点,避免僵尸结点 ev.events = EPOLLIN; epoll_ctl(eventloop->epfd, EPOLL_CTL_DEL, fd, &ev); close(fd); free(si); } else if (ret == 0) { //对端断开连接 printf("fd %d disconnect\n", fd); ev.events = EPOLLIN; epoll_ctl(eventloop->epfd, EPOLL_CTL_DEL, fd, &ev); //close同一断开连接,避免客户端大量的close_wait状态 close(fd); free(si); } else { //打印接收到的数据 printf("recv: %s, %d Bytes\n", si->recvbuffer, ret); //设置sendbuffer si->rlen = ret; memcpy(si->sendbuffer, si->recvbuffer, si->rlen); si->slen = si->rlen; //注册写事件处理器 struct epoll_event ev; ev.events = EPOLLOUT | EPOLLET; si->sockfd = fd; si->callback = send_cb; ev.data.ptr = si; epoll_ctl(eventloop->epfd, EPOLL_CTL_MOD, fd, &ev); } } int accept_cb(int fd, int events, void* arg) { //处理新的连接。 连接IO事件处理流程 struct sockaddr_in cli_addr; memset(&cli_addr, 0, sizeof(cli_addr)); socklen_t cli_len = sizeof(cli_addr); int cli_fd = accept(fd, (SA*)&cli_addr, &cli_len); if (cli_fd <= 0) return -1; char cli_ip[INET_ADDRSTRLEN] = {0}; //存储cli_ip printf("Recv from ip %s at port %d\n", inet_ntop(AF_INET, &cli_addr.sin_addr, cli_ip, sizeof(cli_ip)), ntohs(cli_addr.sin_port)); //注册接下来的读事件处理器 struct epoll_event ev; ev.events = EPOLLIN | EPOLLET; struct sockitem* si = (struct sockitem*)malloc(sizeof(struct sockitem)); si->sockfd = cli_fd; si->callback = recv_cb;//设置事件处理器 ev.data.ptr = si; epoll_ctl(eventloop->epfd, EPOLL_CTL_ADD, cli_fd, &ev); return cli_fd; } int send_cb(int fd, int events, void* arg) { //处理send IO事件 struct sockitem *si = (struct sockitem*)arg; send(fd, si->sendbuffer, si->slen, 0); //再次注册IO读事件处理器 struct epoll_event ev; ev.events = EPOLLIN | EPOLLET; si->sockfd = fd; si->callback = recv_cb;//设置事件处理器 ev.data.ptr = si; epoll_ctl(eventloop->epfd, EPOLL_CTL_MOD, fd, &ev); } int main(int argc, char* argv[]) { if (argc != 2) { fprintf(stderr, "uasge: %s <port>", argv[0]); return 1; } int sockfd = socket(AF_INET, SOCK_STREAM, 0); struct sockaddr_in serv_addr; int port = atoi(argv[1]); //确定服务端协议地址簇 memset(&serv_addr, 0, sizeof(serv_addr)); serv_addr.sin_family = AF_INET; serv_addr.sin_addr.s_addr = INADDR_ANY; serv_addr.sin_port = htons(port); //进行绑定 if (-1 == bind(sockfd, (SA*)&serv_addr, sizeof(serv_addr))) { fprintf(stderr, "bind error"); return 2; } if (-1 == listen(sockfd, 5)) { fprintf(stderr, "listen error"); return 3; } //init eventloop eventloop = (struct reactor*)malloc(sizeof(struct reactor)); //创建epoll句柄. eventloop->epfd = epoll_create(1); //注册建立连接IO事件处理函数 struct epoll_event ev; ev.events = EPOLLIN; struct sockitem* si = (struct sockitem*)malloc(sizeof(struct sockitem)); si->sockfd = sockfd; si->callback = accept_cb;//设置事件处理器 ev.data.ptr = si; //将监视事件加入到reactor的epfd中 epoll_ctl(eventloop->epfd, EPOLL_CTL_ADD, sockfd, &ev); while (1) { //多路复用器监视多个IO事件 int nready = epoll_wait(eventloop->epfd, eventloop->events, 512, -1); if (nready < -1) { break; } int i = 0; //循环分发所有的IO事件给处理器 for (i = 0; i < nready; ++i) { if (eventloop->events[i].events & EPOLLIN) { struct sockitem* si = (struct sockitem*)eventloop->events[i].data.ptr; si->callback(si->sockfd, eventloop->events[i].events, si); } if (eventloop->events[i].events & EPOLLOUT) { struct sockitem* si = (struct sockitem*)eventloop->events[i].data.ptr; si->callback(si->sockfd, eventloop->events[i].events, si); } } } return 0; } Jetbrains全家桶1年46,售后保障稳定
版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。
发布者:全栈程序员栈长,转载请注明出处:https://javaforall.cn/230804.html原文链接:https://javaforall.cn