C++基于reactor的服务器百万并发如何实现(C++,reactor,开发技术)

并发量和承载的概念
并发量：一个服务器能同时承载客户端的数量
承载：客户端发送给服务器的请求(http或tcp等)在200ms内可以返回正确的结果

一、服务器的代码实现与讲解

结构体代码主要构建的结构如图所示
链表结构，每个eventblock结点，包括一个ntyevent数组，数组中存储fd

/*结构体定义链表数组*/structntyevent{ intfd;//要监听的文件描述符 intevents;//对应的监听事件， EPOLLIN和EPOLLOUT（不同的事件，走不同的回调函数） void*arg;//指向自己结构体指针 int(*callback)(intfd,intevents,void*arg); intstatus;//是否在监听：1->在红黑树上（监听），0->不在（不监听） charbuffer[BUFFER_LENGTH]; intlength; longlast_active;};structeventblock{ structeventblock*next; structntyevent*events;//数组};structntyreactor{ //句柄 intepfd; //结点个数 intblkcnt; structeventblock*evblk;//fd-->100w};

初始化fd 上树、下树代码

//nty_event_set(event,sockfd,acceptor,reactor);//初始化sockfdvoidnty_event_set(structntyevent*ev,intfd,NCALLBACKcallback,void*arg){ ev->fd=fd; ev->callback=callback; ev->events=0; ev->arg=arg; ev->last_active=time(NULL); return;}//nty_event_add(reactor->epfd,EPOLLIN,event);//对监听的epoll红黑树上的结点的修改intnty_event_add(intepfd,intevents,structntyevent*ev){ structepoll_eventep_ev={0,{0}}; ep_ev.data.ptr=ev; ep_ev.events=ev->events=events; intop; if(ev->status==1){ op=EPOLL_CTL_MOD; }else{ op=EPOLL_CTL_ADD; ev->status=1; } if(epoll_ctl(epfd,op,ev->fd,&ep_ev)<0){ printf("eventaddfailed[fd=%d],events[%d]\n",ev->fd,events); return-1; } return0;}intnty_event_del(intepfd,structntyevent*ev){ structepoll_eventep_ev={0,{0}}; if(ev->status!=1){ return-1; } ep_ev.data.ptr=ev; ev->status=0; epoll_ctl(epfd,EPOLL_CTL_DEL,ev->fd,&ep_ev); return0;}

回调函数代码的书写
注意看recv_cb的回调函数中，recv之后，立马下树，然后又重新初始化fd，上树。这样做的目的是因为代码逻辑是recv收到数据后，立即原样send,所以需要对fd的属性进行更改，需要重新初始化赋值，然后上树

intrecv_cb(intfd,intevents,void*arg){ structntyreactor*reactor=(structntyreactor*)arg; structntyevent*ev=ntyreactor_idx(reactor,fd); intlen=recv(fd,ev->buffer,BUFFER_LENGTH,0);// nty_event_del(reactor->epfd,ev); if(len>0){ ev->length=len; ev->buffer[len]='\0'; printf("C[%d]:%s\n",fd,ev->buffer); nty_event_set(ev,fd,send_cb,reactor); nty_event_add(reactor->epfd,EPOLLOUT,ev); }elseif(len==0){ close(ev->fd); //printf("[fd=%d]pos[%ld],closed\n",fd,ev-reactor->events); }else{ close(ev->fd); printf("recv[fd=%d]error[%d]:%s\n",fd,errno,strerror(errno)); } returnlen;}intsend_cb(intfd,intevents,void*arg){ structntyreactor*reactor=(structntyreactor*)arg; structntyevent*ev=ntyreactor_idx(reactor,fd); intlen=send(fd,ev->buffer,ev->length,0); if(len>0){ printf("send[fd=%d],[%d]%s\n",fd,len,ev->buffer); nty_event_del(reactor->epfd,ev); nty_event_set(ev,fd,recv_cb,reactor); nty_event_add(reactor->epfd,EPOLLIN,ev); }else{ close(ev->fd); nty_event_del(reactor->epfd,ev); printf("send[fd=%d]error%s\n",fd,strerror(errno)); } returnlen;}intaccept_cb(intfd,intevents,void*arg){ structntyreactor*reactor=(structntyreactor*)arg; if(reactor==NULL)return-1; structsockaddr_inclient_addr; socklen_tlen=sizeof(client_addr); intclientfd; if((clientfd=accept(fd,(structsockaddr*)&client_addr,&len))==-1){ if(errno!=EAGAIN&&errno!=EINTR){ } printf("accept:%s\n",strerror(errno)); return-1; } intflag=0; if((flag=fcntl(clientfd,F_SETFL,O_NONBLOCK))<0){ printf("%s:fcntlnonblockingfailed,%d\n",__func__,MAX_EPOLL_EVENTS); return-1; } /*存储*/ structntyevent*event=ntyreactor_idx(reactor,clientfd); nty_event_set(event,clientfd,recv_cb,reactor); nty_event_add(reactor->epfd,EPOLLIN,event); printf("newconnect[%s:%d],pos[%d]\n", inet_ntoa(client_addr.sin_addr),ntohs(client_addr.sin_port),clientfd); return0;}

链表的初始化与销毁

//初始化链表intntyreactor_init(structntyreactor*reactor){ if(reactor==NULL)return-1; memset(reactor,0,sizeof(structntyreactor)); reactor->epfd=epoll_create(1); if(reactor->epfd<=0){ printf("createepfdin%serr%s\n",__func__,strerror(errno)); return-2; } structntyevent*evs=(structntyevent*)malloc((MAX_EPOLL_EVENTS)*sizeof(structntyevent)); if(evs==NULL){ printf("ntyreactor_allocntyeventsfailed\n"); return-2; } memset(evs,0,(MAX_EPOLL_EVENTS)*sizeof(structntyevent)); structeventblock*block=(structeventblock*)malloc(sizeof(structeventblock)); if(block==NULL){ printf("ntyreactor_alloceventblockfailed\n"); return-2; } memset(block,0,sizeof(structeventblock)); block->events=evs; block->next=NULL; reactor->evblk=block; reactor->blkcnt=1; return0;}

找到fd应在链表数组中存储的位置并返回

//新增块数(eventblock结点个数)//ntyreactor_alloc(reactor);intntyreactor_alloc(structntyreactor*reactor){ if(reactor==NULL)return-1; if(reactor->evblk==NULL)return-1; structeventblock*blk=reactor->evblk; while(blk->next!=NULL){ blk=blk->next; } structntyevent*evs=(structntyevent*)malloc((MAX_EPOLL_EVENTS)*sizeof(structntyevent)); if(evs==NULL){ printf("ntyreactor_allocntyeventsfailed\n"); return-2; } memset(evs,0,(MAX_EPOLL_EVENTS)*sizeof(structntyevent)); structeventblock*block=(structeventblock*)malloc(sizeof(structeventblock)); if(block==NULL){ printf("ntyreactor_alloceventblockfailed\n"); return-2; } memset(block,0,sizeof(structeventblock)); block->events=evs; block->next=NULL; blk->next=block; reactor->blkcnt++;// return0;}//structntyevent*event=ntyreactor_idx(reactor,sockfd);structntyevent*ntyreactor_idx(structntyreactor*reactor,intsockfd){ intblkidx=sockfd/MAX_EPOLL_EVENTS; //如果块数(eventblock结点个数)不能满足新的sockfd的存放 while(blkidx>=reactor->blkcnt){ //新增块数(eventblock结点个数) ntyreactor_alloc(reactor); } //找到存放sockfd的块(eventblock对应的结点) inti=0; structeventblock*blk=reactor->evblk; while(i++<blkidx&&blk!=NULL){ blk=blk->next; } //返回对应块(eventblock对应的结点)的存放sockfd数组的那个具体位置 return&blk->events[sockfd%MAX_EPOLL_EVENTS];}

上树，并初始化链表数组上对应的fd

//ntyreactor_addlistener(reactor,sockfds[i],accept_cb);//上树，并初始化链表数组上对应的fdintntyreactor_addlistener(structntyreactor*reactor,intsockfd,NCALLBACK*acceptor){ if(reactor==NULL)return-1; if(reactor->evblk==NULL)return-1; //reactor->evblk->events[sockfd]; //找到sock所在的具体位置 structntyevent*event=ntyreactor_idx(reactor,sockfd); 初始化sockfd nty_event_set(event,sockfd,acceptor,reactor); //对监听的epoll红黑树上的结点的修改 nty_event_add(reactor->epfd,EPOLLIN,event); return0;}

epollwait

//ntyreactor_run(reactor);intntyreactor_run(structntyreactor*reactor){ if(reactor==NULL)return-1; if(reactor->epfd<0)return-1; if(reactor->evblk==NULL)return-1; structepoll_eventevents[MAX_EPOLL_EVENTS+1]; intcheckpos=0,i; while(1){/* longnow=time(NULL); for(i=0;i<100;i++,checkpos++){ if(checkpos==MAX_EPOLL_EVENTS){ checkpos=0; } if(reactor->events[checkpos].status!=1){ continue; } longduration=now-reactor->events[checkpos].last_active; if(duration>=60){ close(reactor->events[checkpos].fd); printf("[fd=%d]timeout\n",reactor->events[checkpos].fd); nty_event_del(reactor->epfd,&reactor->events[checkpos]); } }*/ intnready=epoll_wait(reactor->epfd,events,MAX_EPOLL_EVENTS,1000); if(nready<0){ printf("epoll_waiterror,exit\n"); continue; } for(i=0;i<nready;i++){ structntyevent*ev=(structntyevent*)events[i].data.ptr; //看fd连接是否发生变化 if((events[i].events&EPOLLIN)&&(ev->events&EPOLLIN)){ ev->callback(ev->fd,events[i].events,ev->arg); } if((events[i].events&EPOLLOUT)&&(ev->events&EPOLLOUT)){ ev->callback(ev->fd,events[i].events,ev->arg); } } }}

main函数；此服务器代码开设了100个监听的端口，目的是因为客户端测试程序也是运行在虚拟机的Ubuntu上，通过开三台来充当客户端来进行测试。有因为一台Ubuntu最多有6w个端口，3台有18w端口。如果服务器只开设一个监听端口，则最多有18w端口。因此要达到100w并发则应多开设端口

//3,6w,1,100==//<remoteip,remoteport,localip,localport>intmain(intargc,char*argv[]){ unsignedshortport=SERVER_PORT;//listen8888 if(argc==2){ port=atoi(argv[1]);//把参数str所指向的字符串转换为一个整数（类型为int型） } structntyreactor*reactor=(structntyreactor*)malloc(sizeof(structntyreactor)); /*初始化三个结构体，建立链表*/ ntyreactor_init(reactor); inti=0; intsockfds[PORT_COUNT]={0}; for(i=0;i<PORT_COUNT;i++){ //端口号的监听 sockfds[i]=init_sock(port+i); //上树 ntyreactor_addlistener(reactor,sockfds[i],accept_cb); } //epoll_wait ntyreactor_run(reactor); // ntyreactor_destory(reactor); for(i=0;i<PORT_COUNT;i++){ close(sockfds[i]); } free(reactor); return0;}

完整服务器代码展示

/*链表存储数组，把epoll变成对事件的管理,用链表数组的目的就是为了回调函数*//*recv写法：代码逻辑是收到数据后，立即原样返回所以才那样写*/#include<stdio.h>#include<stdlib.h>#include<string.h>#include<sys/socket.h>#include<sys/epoll.h>#include<arpa/inet.h>#include<fcntl.h>#include<unistd.h>#include<errno.h>#include<time.h>#defineBUFFER_LENGTH 4096#defineMAX_EPOLL_EVENTS 1024#defineSERVER_PORT 8888#definePORT_COUNT 100typedefintNCALLBACK(int,int,void*);//structntyevent*evs=(structntyevent*)malloc((MAX_EPOLL_EVENTS)*sizeof(structntyevent));structntyevent{ intfd;//要监听的文件描述符 intevents;//对应的监听事件， EPOLLIN和EPOLLOUT（不同的事件，走不同的回调函数） void*arg;//指向自己结构体指针 int(*callback)(intfd,intevents,void*arg); intstatus;//是否在监听：1->在红黑树上（监听），0->不在（不监听） charbuffer[BUFFER_LENGTH]; intlength; longlast_active;};structeventblock{ structeventblock*next; structntyevent*events;//数组};structntyreactor{ //句柄 intepfd; //结点个数 intblkcnt; structeventblock*evblk;//fd-->100w};intrecv_cb(intfd,intevents,void*arg);intsend_cb(intfd,intevents,void*arg);structntyevent*ntyreactor_idx(structntyreactor*reactor,intsockfd);//nty_event_set(event,sockfd,acceptor,reactor);//初始化sockfdvoidnty_event_set(structntyevent*ev,intfd,NCALLBACKcallback,void*arg){ ev->fd=fd; ev->callback=callback; ev->events=0; ev->arg=arg; ev->last_active=time(NULL); return;}//nty_event_add(reactor->epfd,EPOLLIN,event);//对监听的epoll红黑树上的结点的修改intnty_event_add(intepfd,intevents,structntyevent*ev){ structepoll_eventep_ev={0,{0}}; ep_ev.data.ptr=ev; ep_ev.events=ev->events=events; intop; if(ev->status==1){ op=EPOLL_CTL_MOD; }else{ op=EPOLL_CTL_ADD; ev->status=1; } if(epoll_ctl(epfd,op,ev->fd,&ep_ev)<0){ printf("eventaddfailed[fd=%d],events[%d]\n",ev->fd,events); return-1; } return0;}intnty_event_del(intepfd,structntyevent*ev){ structepoll_eventep_ev={0,{0}}; if(ev->status!=1){ return-1; } ep_ev.data.ptr=ev; ev->status=0; epoll_ctl(epfd,EPOLL_CTL_DEL,ev->fd,&ep_ev); return0;}intrecv_cb(intfd,intevents,void*arg){ structntyreactor*reactor=(structntyreactor*)arg; structntyevent*ev=ntyreactor_idx(reactor,fd); intlen=recv(fd,ev->buffer,BUFFER_LENGTH,0);// nty_event_del(reactor->epfd,ev); if(len>0){ ev->length=len; ev->buffer[len]='\0'; printf("C[%d]:%s\n",fd,ev->buffer); nty_event_set(ev,fd,send_cb,reactor); nty_event_add(reactor->epfd,EPOLLOUT,ev); }elseif(len==0){ close(ev->fd); //printf("[fd=%d]pos[%ld],closed\n",fd,ev-reactor->events); }else{ close(ev->fd); printf("recv[fd=%d]error[%d]:%s\n",fd,errno,strerror(errno)); } returnlen;}intsend_cb(intfd,intevents,void*arg){ structntyreactor*reactor=(structntyreactor*)arg; structntyevent*ev=ntyreactor_idx(reactor,fd); intlen=send(fd,ev->buffer,ev->length,0); if(len>0){ printf("send[fd=%d],[%d]%s\n",fd,len,ev->buffer); nty_event_del(reactor->epfd,ev); nty_event_set(ev,fd,recv_cb,reactor); nty_event_add(reactor->epfd,EPOLLIN,ev); }else{ close(ev->fd); nty_event_del(reactor->epfd,ev); printf("send[fd=%d]error%s\n",fd,strerror(errno)); } returnlen;}intaccept_cb(intfd,intevents,void*arg){ structntyreactor*reactor=(structntyreactor*)arg; if(reactor==NULL)return-1; structsockaddr_inclient_addr; socklen_tlen=sizeof(client_addr); intclientfd; if((clientfd=accept(fd,(structsockaddr*)&client_addr,&len))==-1){ if(errno!=EAGAIN&&errno!=EINTR){ } printf("accept:%s\n",strerror(errno)); return-1; } intflag=0; if((flag=fcntl(clientfd,F_SETFL,O_NONBLOCK))<0){ printf("%s:fcntlnonblockingfailed,%d\n",__func__,MAX_EPOLL_EVENTS); return-1; } /*存储*/ structntyevent*event=ntyreactor_idx(reactor,clientfd); nty_event_set(event,clientfd,recv_cb,reactor); nty_event_add(reactor->epfd,EPOLLIN,event); printf("newconnect[%s:%d],pos[%d]\n", inet_ntoa(client_addr.sin_addr),ntohs(client_addr.sin_port),clientfd); return0;}intinit_sock(shortport){ intfd=socket(AF_INET,SOCK_STREAM,0); fcntl(fd,F_SETFL,O_NONBLOCK); structsockaddr_inserver_addr; memset(&server_addr,0,sizeof(server_addr)); server_addr.sin_family=AF_INET; server_addr.sin_addr.s_addr=htonl(INADDR_ANY); server_addr.sin_port=htons(port); bind(fd,(structsockaddr*)&server_addr,sizeof(server_addr)); if(listen(fd,20)<0){ printf("listenfailed:%s\n",strerror(errno)); } returnfd;}//新增块数(eventblock结点个数)//ntyreactor_alloc(reactor);intntyreactor_alloc(structntyreactor*reactor){ if(reactor==NULL)return-1; if(reactor->evblk==NULL)return-1; structeventblock*blk=reactor->evblk; while(blk->next!=NULL){ blk=blk->next; } structntyevent*evs=(structntyevent*)malloc((MAX_EPOLL_EVENTS)*sizeof(structntyevent)); if(evs==NULL){ printf("ntyreactor_allocntyeventsfailed\n"); return-2; } memset(evs,0,(MAX_EPOLL_EVENTS)*sizeof(structntyevent)); structeventblock*block=(structeventblock*)malloc(sizeof(structeventblock)); if(block==NULL){ printf("ntyreactor_alloceventblockfailed\n"); return-2; } memset(block,0,sizeof(structeventblock)); block->events=evs; block->next=NULL; blk->next=block; reactor->blkcnt++;// return0;}//structntyevent*event=ntyreactor_idx(reactor,sockfd);structntyevent*ntyreactor_idx(structntyreactor*reactor,intsockfd){ intblkidx=sockfd/MAX_EPOLL_EVENTS; //如果块数(eventblock结点个数)不能满足新的sockfd的存放 while(blkidx>=reactor->blkcnt){ //新增块数(eventblock结点个数) ntyreactor_alloc(reactor); } //找到存放sockfd的块(eventblock对应的结点) inti=0; structeventblock*blk=reactor->evblk; while(i++<blkidx&&blk!=NULL){ blk=blk->next; } //返回对应块(eventblock对应的结点)的存放sockfd数组的那个具体位置 return&blk->events[sockfd%MAX_EPOLL_EVENTS];}//初始化链表intntyreactor_init(structntyreactor*reactor){ if(reactor==NULL)return-1; memset(reactor,0,sizeof(structntyreactor)); reactor->epfd=epoll_create(1); if(reactor->epfd<=0){ printf("createepfdin%serr%s\n",__func__,strerror(errno)); return-2; } structntyevent*evs=(structntyevent*)malloc((MAX_EPOLL_EVENTS)*sizeof(structntyevent)); if(evs==NULL){ printf("ntyreactor_allocntyeventsfailed\n"); return-2; } memset(evs,0,(MAX_EPOLL_EVENTS)*sizeof(structntyevent)); structeventblock*block=(structeventblock*)malloc(sizeof(structeventblock)); if(block==NULL){ printf("ntyreactor_alloceventblockfailed\n"); return-2; } memset(block,0,sizeof(structeventblock)); block->events=evs; block->next=NULL; reactor->evblk=block; reactor->blkcnt=1; return0;}intntyreactor_destory(structntyreactor*reactor){ close(reactor->epfd); //free(reactor->events); structeventblock*blk=reactor->evblk; structeventblock*blk_next=NULL; while(blk!=NULL){ blk_next=blk->next; free(blk->events); free(blk); blk=blk_next; } return0;}//ntyreactor_addlistener(reactor,sockfds[i],accept_cb);//上树，并初始化链表数组上对应的fdintntyreactor_addlistener(structntyreactor*reactor,intsockfd,NCALLBACK*acceptor){ if(reactor==NULL)return-1; if(reactor->evblk==NULL)return-1; //reactor->evblk->events[sockfd]; //找到sock所在的具体位置 structntyevent*event=ntyreactor_idx(reactor,sockfd); 初始化sockfd nty_event_set(event,sockfd,acceptor,reactor); //对监听的epoll红黑树上的结点的修改 nty_event_add(reactor->epfd,EPOLLIN,event); return0;}//ntyreactor_run(reactor);intntyreactor_run(structntyreactor*reactor){ if(reactor==NULL)return-1; if(reactor->epfd<0)return-1; if(reactor->evblk==NULL)return-1; structepoll_eventevents[MAX_EPOLL_EVENTS+1]; intcheckpos=0,i; while(1){/* longnow=time(NULL); for(i=0;i<100;i++,checkpos++){ if(checkpos==MAX_EPOLL_EVENTS){ checkpos=0; } if(reactor->events[checkpos].status!=1){ continue; } longduration=now-reactor->events[checkpos].last_active; if(duration>=60){ close(reactor->events[checkpos].fd); printf("[fd=%d]timeout\n",reactor->events[checkpos].fd); nty_event_del(reactor->epfd,&reactor->events[checkpos]); } }*/ intnready=epoll_wait(reactor->epfd,events,MAX_EPOLL_EVENTS,1000); if(nready<0){ printf("epoll_waiterror,exit\n"); continue; } for(i=0;i<nready;i++){ structntyevent*ev=(structntyevent*)events[i].data.ptr; //看fd连接是否发生变化 if((events[i].events&EPOLLIN)&&(ev->events&EPOLLIN)){ ev->callback(ev->fd,events[i].events,ev->arg); } if((events[i].events&EPOLLOUT)&&(ev->events&EPOLLOUT)){ ev->callback(ev->fd,events[i].events,ev->arg); } } }}//3,6w,1,100==//<remoteip,remoteport,localip,localport>intmain(intargc,char*argv[]){ unsignedshortport=SERVER_PORT;//listen8888 if(argc==2){ port=atoi(argv[1]);//把参数str所指向的字符串转换为一个整数（类型为int型） } structntyreactor*reactor=(structntyreactor*)malloc(sizeof(structntyreactor)); /*初始化三个结构体，建立链表*/ ntyreactor_init(reactor); inti=0; intsockfds[PORT_COUNT]={0}; for(i=0;i<PORT_COUNT;i++){ //端口号的监听 sockfds[i]=init_sock(port+i); //上树 ntyreactor_addlistener(reactor,sockfds[i],accept_cb); } //epoll_wait ntyreactor_run(reactor); // ntyreactor_destory(reactor); for(i=0;i<PORT_COUNT;i++){ close(sockfds[i]); } free(reactor); return0;}

reactor的写法感觉和epoll的普通写法，感觉差别就是reactor多了个回调函数，具体没啥优点？
epoll是针对io的管理。 reactor对针对事件的管理
不同的事件，针对不同的回调函数
性能上没啥差异，但提高了代码的复用性。具体需要自己慢慢体会体会，呜呜呜呜还有体会到，编程思想不过关。

二、环境设置

限制是fd的限制，系统默认fd最多有1024个，按照一个连接一个fd的做法，那就需要百万个fd。这里有两种修改方法，一是使用ulimit -n命令，这个命令重启就失效；二是修改/etc/security/limits.conf文件，这是永久有效的，重启或sysctl -p生效。

*hardnofile1048576*softnofile1048576

hard是硬限制，不能超过该值，soft是软限制，可以超过，超过后就开始回收。
这个文件里还有一些其他的参数可以了解一下，fs.file_max是fd可取到的最大值，注意与fd最大个数区分。
突破这两个限制后，还会遇到一个问题，客户端会报错：connection timedout。连接超时，即是客户端未收到服务器对客户端connect()的回应包。这里有两种可能，客户端为收到服务器的包或是服务器未收到客户端的connect包。事实上，是因为系统有个防火墙iotables，这个防火墙是基于网卡和协议栈之间的过滤机制netfilter实现的。netfilter当连接数到达一定程度时，会不允许再向外发送connect包。修改也是通过/etc/security/limits.conf文件

net.nf_conntrack_max=1048576

突破这些限制，就可以实现百万并发了。
这里再介绍/etc/security/limits.conf中几个参数
net.ipv4.tcp_mem=262144 524288 786432是所有TCP协议栈所占空间的大小，单位是页（4KB）。介绍一下后面写的三个值，当所占空间大小超过第二个值时，系统会进行优化，此时如果占用空间降到第一个值以下，不再优化，第三个值是上限，不允许分配超过比大小的空间。
net.ipv4.tcp_wmem=2048 2048 4096是每个socket对应的写缓冲区大小，三个值分别是最小值、默认值、最大值，单位是B。
net.ipv4.tcp_rmem=2048 2048 4096是每个socket对应的读缓冲区大小，三个值分别是最小值、默认值、最大值，单位是B。
做百万并发时，如果内存不大，可以相应调小。在实际应用中，如果传输大文件，调大；如果传输的都是字符，调小，就可以接收更多fd。