百万级高并发，可以使用epoll模型，那么io是存放在哪里的？
reactor反应堆
如何实现reactor反应堆？
1、将epoll 里面的客户端fd 节点，做成event，event里面携带io的相关数据
2、accept的时候，只关心，读取数据和发送数据
一般情况下，客户端连接上服务器之后，会先发数据给服务器，因此连接后，服务器这边的做法先是 读取客户端套接字里面的信息，再是做send操作
编码
写一个demo
#include <stdio.h> #include <stdlib.h> #include <string.h> #include <errno.h> #include <time.h> #include <unistd.h> #include <sys/types.h> #include <sys/socket.h> #include <arpa/inet.h> #include <fcntl.h> #include <sys/epoll.h> //1、定义数据结构 #define BUFFER_LENGTH 1024 #define MAX_EPOLL_EVENTS 1024 typedef int (*QBCALLBACK)(int fd,int events, void *arg); //用于用户层的event struct qbevent{ int fd; int events; void *arg; int (*callback)(int fd,int events, void *arg); int status; char buff[BUFFER_LENGTH]; int length; long last_alive; }; struct qbreactor{ int epollfd; struct qbevent * events; }; //回调函数，发送数据和 接受数据的 int send_cb(int fd,int events, void *arg); int recv_cb(int fd,int events, void *arg); //2、写event的 set，add，del //将传入进来的qbevent指针对应的参数进行初始化 void qb_event_set(struct qbevent *ev,int fd,QBCALLBACK call,void * arg) { ev->fd = fd; ev->arg = arg; ev->events = 0; ev->callback = call; ev->status = 0; ev->last_alive= time(NULL); return ; } //将用户的event 挂到 内核epoll的event上 int qb_event_add(int epollfd,int event,struct qbevent *ev) { if(epollfd < 0) return -1; if(ev == NULL) return -1; struct epoll_event epev={0,{0}}; epev.data.ptr = ev; epev.events = ev->events = event; //用户event 和 内核的event 分别赋值 int opt; if(ev->status == 1) { opt = EPOLL_CTL_MOD; } else { opt = EPOLL_CTL_ADD; ev->status = 1; } if(epoll_ctl(epollfd,opt,ev->fd,&epev) < 0) { printf("event add failed [fd=%d], events[%d]\n", ev->fd,ev->events); return -1; } return 0; } //3、从内核epol l event 中删除 客户端的fd，此处为用户的event int qb_event_del(int epollfd,struct qbevent *ev) { if(epollfd < 0) return -1; if(ev == NULL) return -1; if(ev->status != 1) return -1; struct epoll_event epev={0,{0}};; epev.data.ptr = ev; ev->status = 0; epoll_ctl(epollfd,EPOLL_CTL_DEL,ev->fd,&epev); return 0; } //回调函数 发送数据 int send_cb(int fd,int events, void *arg) { struct qbreactor *reactor = (struct qbreactor *)arg; struct qbevent *ev = reactor->events +fd; int len = send(ev->fd,ev->buff,ev->length,0); if(len > 0){ //说明数据发送ok printf("send[fd=%d], [%d]%s\n", fd, len, ev->buff); qb_event_del(reactor->epollfd,ev); qb_event_set(ev,ev->fd,recv_cb,reactor); qb_event_add(reactor->epollfd,EPOLLIN,ev); } else{ //len < 0 可能写情况是，缓冲区里面需要send的数据是有的， //但是被别的事件提前send出去了，因此本次的send会返回-1 close(ev->fd); qb_event_del(reactor->epollfd,ev); printf("send [fd=%d] error %s\n", fd, strerror(errno)); } return 0; } //回调函数 接收数据 int recv_cb(int fd,int events, void *arg) { struct qbreactor *reactor = (struct qbreactor *)arg; struct qbevent *ev = reactor->events +fd; int len = recv(ev->fd,ev->buff,BUFFER_LENGTH,0); qb_event_del(reactor->epollfd,ev); if(len >0){ ev->buff[len] = '\0'; ev->length = len; //可以进行判断包是否完整，黏包，拆包的事情在这里做 //... printf("recv [%d]:%s\n", fd, ev->buff); //服务器读完了客户端的数据后，想客户端发送数据 qb_event_set(ev,ev->fd, send_cb, reactor); qb_event_add(reactor->epollfd,EPOLLOUT,ev); } else if(len < 0){ close(ev->fd); printf("recv[fd=%d] error[%d]:%s\n", fd, errno, strerror(errno)); } else{ close(ev->fd); printf("[fd=%d] pos[%ld], closed\n", fd, ev-reactor->events); } return 0; } //回调函数 监听新的连接 int accept_cb(int fd,int events, void *arg) { struct qbreactor * reactor = (struct qbreactor *)arg; if (reactor == NULL) return -1; //初始化客户端的地址结构 struct sockaddr_in clientaddr; socklen_t len = sizeof(clientaddr); int clientfd; if ((clientfd = accept(fd, (struct sockaddr*)&clientaddr, &len)) == -1) { if (errno != EAGAIN && errno != EINTR) { //数据被别的事件读取了 } printf("accept: %s\n", strerror(errno)); return -1; } //开始 把clientfd 放到用户层事件里面 int i = 0; do{ //找到clientfd 需要加入的位置 for(i = 0;i<MAX_EPOLL_EVENTS;i++){ if(reactor->events[i].status == 0) break; } if(i == MAX_EPOLL_EVENTS){ printf("%s: max connect limit[%d]\n", __func__, MAX_EPOLL_EVENTS); break; } int flag = 0; if ((flag = fcntl(clientfd, F_SETFL, O_NONBLOCK)) < 0) { printf("%s: fcntl nonblocking failed, %d\n", __func__, MAX_EPOLL_EVENTS); break; } //printf(" ii == %d -- client fd == %d , sockfd = %d\n",i,clientfd,fd); qb_event_set(&reactor->events[clientfd],clientfd,recv_cb,reactor); qb_event_add(reactor->epollfd,EPOLLIN,&reactor->events[clientfd]); }while(0); printf("new connect [%s:%d][time:%ld], pos[%d]\n", inet_ntoa(clientaddr.sin_addr), ntohs(clientaddr.sin_port), reactor->events[i].last_alive, i); return 0; } //4、sock_init 初始化 int sock_init(int port) { int sockfd = socket(AF_INET,SOCK_STREAM,0); if(sockfd < 0){ perror("sockfd socket error\n"); return -1; } //设置为非阻塞 fcntl(sockfd,F_SETFL,O_NONBLOCK); //初始化属性 struct sockaddr_in seraddr; memset(&seraddr,0,sizeof(seraddr)); seraddr.sin_family = AF_INET; seraddr.sin_port = htons(port); seraddr.sin_addr.s_addr = INADDR_ANY; //绑定属性 if(bind(sockfd,(struct sockaddr*)&seraddr,sizeof(seraddr)) < 0){ perror("bind error"); return -1; } //设置最大监听数量 if(listen(sockfd,20)<0){ perror("listen error"); return -1; } return sockfd; } //5、qbreactor的初始化 int qbreactor_init(struct qbreactor *reactor) { if(reactor == NULL) return -1; memset(reactor,0,sizeof(struct qbreactor)); reactor->epollfd = epoll_create(1); if(reactor->epollfd <= 0 ){ printf("create epfd in %s err %s\n", __func__, strerror(errno)); return -1; } reactor->events = (struct qbevent *)malloc(sizeof(struct qbevent )* MAX_EPOLL_EVENTS); if(reactor->events == NULL ){ printf("malloc events in %s err %s\n", __func__, strerror(errno)); close(reactor->epollfd); return -1; } return 0; } //6、qbreactor destroy void qbreactor_destroy(struct qbreactor *reactor) { if(reactor == NULL) return ; if(reactor->events == NULL) return ; close(reactor->epollfd); free(reactor->events); return ; } //7、qbreactor的listen + 传参回调 //添加listen int qbreactor_addlistener(struct qbreactor * reactor,int sockfd,QBCALLBACK accept) { if(reactor == NULL) return -1; if(sockfd < 0) return -1; if(accept == NULL) return -1; qb_event_set(&reactor->events[sockfd],sockfd,accept,reactor); qb_event_add(reactor->epollfd,EPOLLIN,&reactor->events[sockfd]);//作为用来连接的，使用水平触发 return 0; } //8、 run qbreactor void qbreactor_runner(struct qbreactor *reactor) { if(reactor == NULL) return ; if(reactor->epollfd <= 0) return ; if(reactor->events == NULL) return ; //阻塞等待事件发生 struct epoll_event epev[MAX_EPOLL_EVENTS+1]={0}; while(1){ int nready = epoll_wait(reactor->epollfd,epev,MAX_EPOLL_EVENTS,-1); for(int i = 0;i<nready;i++){ struct qbevent * ev = (struct qbevent *) epev[i].data.ptr; //只关心读事件和写事件 //读事件 EPOLLIN if((ev->events & EPOLLIN) && (epev[i].events & EPOLLIN )){ ev->callback(ev->fd,ev->events,ev->arg); } //写事件 EPOLLOUT if((ev->events & EPOLLOUT) && (epev[i].events & EPOLLOUT )){ ev->callback(ev->fd,ev->events,ev->arg); } } } return ; } int main(int argc,char * argv[]) { //串口号从命令行传入 if(argc < 2){ perror("please input port number\n"); return -1; } //获取用于连接的套接字 int sockfd = sock_init(atoi(argv[1])); //初始化reactor struct qbreactor *reactor = (struct qbreactor *)malloc(sizeof(struct qbreactor)); qbreactor_init(reactor); //注册监听 qbreactor_addlistener(reactor,sockfd,accept_cb); //开始运行 qbreactor_runner(reactor); //摧毁reactor qbreactor_destroy(reactor); close(sockfd); return 0; }
