怎样才能更好的封装epoll
从epoll到reactor
既然epoll已经能够从底层支撑了大并发网络服务器的实现，接下来要做的就是怎么将epoll更好的封装，以使其更好的适配上层应用，上层往往不关注底层的连接和IO,更关注对不同就绪事件的处理，故而诞生了事件驱动的reactor
接下来梳理一遍reactor的工作流程，并展示它对各种功能接口的封装细节
int main(){ unsigned short port = 2000; //创建epoll epfd = epoll_create(1); int i = 0; //开放监听MAX_PORT个端口 for(i=0; i<MAX_PORT; i++){ int sockfd = init_server(port+i); conn_list[sockfd].fd = sockfd; conn_list[sockfd].r_action.accept_callback = accept_cb; set_event(sockfd, EPOLLIN,1); } //初始化计时器 gettimeofday(&begin,NULL); //mainloop while (1){ //创建用于接受就绪事件的集合 struct epoll_event events[1024]={0}; //等待就绪事件 int nready=epoll_wait(epfd, events, 1024, -1); //逐个处理就绪事件，reactor的特点，按就绪事件类型来处理 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); } } } }
我们选择使用数组存储连接标识符fd，同时创建了一个结构体来存存储fd的所有信息
//连接标识符，读缓冲区，写缓冲区，三个回调函数 struct conn { int fd; char rbuffer[BUFFER_LENGTH]; int rlength; char wbuffer[BUFFER_LENGTH]; int wlength; RCALLBACK send_callback; //union一段内存，两个名字 union{ RCALLBACK recv_callback; RCALLBACK accept_callback; }r_action; }; struct conn conn_list[CONN_SIZE] = {0};
封装对服务器的初始化功能
//初始化服务器，创建套接字并向内核申请把本机ip与端口同套接字绑定，设置套接字状态为监听，为后续连接做准备 int init_server(int port){ int sockfd = socket(AF_INET, SOCK_STREAM, 0); struct sockaddr_in servaddr; servaddr.sin_family = AF_INET; servaddr.sin_addr.s_addr = htonl(INADDR_ANY); // 0.0.0.0 servaddr.sin_port = htons(port); // 0-1023, if (-1 == bind(sockfd, (struct sockaddr*)&servaddr, sizeof(struct sockaddr))) { printf("bind failed: %s\n", strerror(errno)); } listen(sockfd, 10); printf("listen finshed: %d\n", sockfd); // 3 return sockfd; }
封装更改标识符设置功能
//连接标识符属性设置，可以实现监听和监听事件的修改 int set_event(int fd, int event,int flag){ if(fd < 0 || fd >= CONN_SIZE){ return -1; } //flag用来区别设置监听还是修改监听事件 if(flag){ 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); } }
封装注册函数，初始化struct conn结构体实例
//注册，实现fd的回调函数，缓存空间的初始化 int event_register(int fd, int event){ if(fd < 0 || fd >= CONN_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); }
回调函数——建立新连接
//建立新连接，在监听套接字监听到客户端发起连接 int accept_cb(int fd){ if(fd < 0 || fd >= CONN_SIZE){ return -1; } struct sockaddr_in clientaddr; socklen_t len = sizeof(clientaddr); int clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len); if(clientfd < 0) { return -1; } //每建立1000个连接打印一次总连接数，打印建立1000连接所需时间 if(clientfd % 1000 == 0){ struct timeval current; gettimeofday(¤t,NULL); int time_used = TIME_SUB_MS(current,begin); memcpy(&begin,¤t,sizeof(struct timeval)); printf("accept finished: %d, timeused: %d\n",clientfd,time_used); } event_register(clientfd, EPOLLIN); return 0; }
回调函数——接收数据
//接收数据，并将关注事件改为EPOLLOUT int recv_cb(int fd){ if(fd < 0 || fd >= CONN_SIZE){ return -1; } //将数据接收到rbuffer memset(conn_list[fd].rbuffer, 0, BUFFER_LENGTH ); int count = recv(fd, conn_list[fd].rbuffer, BUFFER_LENGTH, 0); conn_list[fd].rlength = count; //disconnect if (count == 0) { printf("client disconnect: %d\n", fd); close(fd); epoll_ctl(epfd, EPOLL_CTL_DEL, fd, NULL); return 0; } //printf("RECV: %s\n", conn_list[fd].rbuffer); //将数据从rbuffer copy 到 wbuffer 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; }
回调函数——发送数据
//将数据发送，并将fd关注事件调整为EPOLLIN int send_cb(int fd){ if(fd < 0 || fd >= CONN_SIZE){ return -1; } int count = send(fd, conn_list[fd].wbuffer, conn_list[fd].wlength, 0); set_event(fd, EPOLLIN,0); return count; }
