reactor介绍以及reactor实现自发自送TCP服务器

Reactor 的概念Reactor 是一种设计模式用于处理并发事件驱动的系统。它通过将事件分派给对应的处理程序Handler来响应多个并发请求通常用于网络服务器、高性能I/O框架等场景。核心思想是使用单线程或有限线程监听多个事件源如Socket连接通过事件循环Event Loop实现非阻塞处理。采用Reactor将并发请求从I/O管理转变为事件管理。主要通过采用不同的I/O事件对应不同的回调函数来实现。Reactor 的工作流程注册事件将事件源如Socket及其对应的处理器注册到事件多路复用器。事件循环循环调用事件多路复用器如epoll_wait监听就绪事件。事件分发将就绪事件分发给绑定的处理器执行具体逻辑如读取数据、处理请求。C通过Reactor实现TCP“回声”服务器以端口号8000为例通过reactor实现一个简单的TCP服务器实现“回声”功能将客户端发来的数据原样返回。#define PORT 8000 #define BUFFERLENGTH 1024 #define CONN_SIZE 1048576 typedef int(*RCALLBACK)(int fd); int epfd 0; int accept_cb(int fd); int send_cb(int fd); int recv_cb(int fd); struct conn{ int fd; char rbuffer[BUFFERLENGTH]; int rlength; char wbuffer[BUFFERLENGTH]; int wlength; // POLLOUT RCALLBACK send_callback; // POLLIN union{ RCALLBACK accept_callback; RCALLBACK recv_callback; }r_action; }; struct conn conn_list[CONN_SIZE] {0}; // 定义事件设定 int set_event(int fd, int event, int flag){ if(flag 1){ struct epoll_event ev; ev.events event; ev.data.fd fd; 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 event_register(int fd, int event){ if(fd 0) return -1; conn_list[fd].fd fd; conn_list[fd].send_callback send_cb; conn_list[fd].r_action.recv_callback recv_cb; memset(conn_list[fd].rbuffer, 0, BUFFERLENGTH); conn_list[fd].rlength 0; memset(conn_list[fd].wbuffer, 0, BUFFERLENGTH); conn_list[fd].wlength 0; set_event(fd, event, 1); return 0; } // 设定接收客户端的回调函数 int accept_cb(int fd){ struct sockaddr_in clientaddr; socklen_t len sizeof(clientaddr); int clientfd accept(fd, (struct*) clientaddr, len); if(clientfd0) return -1; event_register(clientfd, EPOLLIN); return 0; } int recv_cb(int fd){ int count recv(fd, conn_list[fd].rbuffer, BUFFERLENGTH, 0); if(count 0){ close(fd); epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); return 0; } conn_list[fd].rlength 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 0; } int send_cb(int fd){ memset(conn_list[fd].rbuffer, 0, BUFFERLENGTH); int count send(fd, conn_list[fd].wbuffer, conn_list[fd].wlength, 0); set_event(fd, EPOLLIN, 0); return count; } int init_server(unsigned short port){ int sockfd socket(AF_INET, SOCK_STREAM, 0); struct sockaddr_in serveraddr; serveraddr.sin_family AF_INET; serveraddr.sin_addr.s_addr htonl(INADDY_ANY); serveraddr.sin_port htons(port); if(-1 bind(sockfd, (struct sockaddr*)serveraddr, sizeof(struct sockaddr))){ printf(bind error:%s\n, strerror(errno)); return -1; } listen(sockfd, 10); return sockfd; } int main(){ epfd epoll_create(1); int sockfd init_server(PORT); conn_list[sockfd].fd sockfd; conn_list[sockfd].r_action.recv_callback accept_cb; set_event(sockfd, EPOLLIN, 0); while(1){ struct epoll_event events[1024] {0}; int nready epoll_wait(epfd, events, 1024, -1); int i; 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); } } } }首先定义reactor结构体包括fd、event、写入的参数rbuffer、rlength读取的参数wbuffer、rlength。以及包括接收新的客户端accept_callback发送数据的回调函数send_callback以及接收数据的回调函数recv_callback。首先创建epoll实例对目标端口进行监听。之后通过epoll_wait等待事件就绪从而实现TCP服务器自发自送。Reactor写法的优点高可扩展性通过回调函数如accept_cb、recv_cb分离协议处理新增协议或修改逻辑只需替换对应回调无需改动事件循环主体。连接状态管理每个连接独立持有状态缓冲区、数据长度、回调指针便于处理部分收发、连接级超时、多步协议等复杂场景避免数据丢失。统一的事件驱动模型所有 I/O 事件接受连接、读写、错误集中处理易于添加定时器、信号等额外事件源。