您的位置: 首页  >  文章  >  高并发的epoll+线程池，epoll在线程池内

高并发的epoll+线程池，epoll在线程池内

分类: 文章 • 2025-04-11 10:01:10

文章来源：http://blog.chinaunix.net/uid-311680-id-2439723.html

我们知道，服务器并发模型通常可分为单线程和多线程模型，这里的线程通常是指“I/O线程”，即负责I/O操作，协调分配任务的“管理线程”，而实际的请求和任务通常交由所谓“工作者线程”处理。通常多线程模型下，每个线程既是I/O线程又是工作者线程。所以这里讨论的是，单I/O线程+多工作者线程的模型，这也是最常用的一种服务器并发模型。我所在的项目中的server代码中，这种模型随处可见。它还有个名字，叫“半同步/半异步“模型，同时，这种模型也是生产者/消费者（尤其是多消费者）模型的一种表现。

这种架构主要是基于I/O多路复用的思想（主要是epoll，select/poll已过时），通过单线程I/O多路复用，可以达到高效并发，同时避免了多线程I/O来回切换的各种开销，思路清晰，易于管理，而基于线程池的多工作者线程，又可以充分发挥和利用多线程的优势，利用线程池，进一步提高资源复用性和避免产生过多线程。

 

1 模型架构

高并发的epoll+线程池，epoll在线程池内 

 

2 实现要点

2.1 单I/O 线程epoll

实现单I/O线程的epoll模型是本架构的第一个技术要点，主要思想如下： 

单线程创建epoll并等待，有I/O请求（socket）到达时，将其加入epoll并从线程池中取一个空闲工作者线程，将实际的任务交由工作者线程处理。

伪码：

高并发的epoll+线程池，epoll在线程池内
创建一个epoll实例;
while(server running)
{
    epoll等待事件;
    if(新连接到达且是有效连接)
    {
        accept此连接;
        将此连接设置为non-blocking;
        为此连接设置event(EPOLLIN | EPOLLET ...);
        将此连接加入epoll监听队列;
        从线程池取一个空闲工作者线程并处理此连接;
    }
    else if(读请求)
    {
        从线程池取一个空闲工作者线程并处理读请求;
    }
    else if(写请求)
    {
        从线程池取一个空闲工作者线程并处理写请求;
    }
    else
        其他事件;     
}
高并发的epoll+线程池，epoll在线程池内

伪码可能写的不太好，其实就是基本的epoll使用。

但要注意和线程池的配合使用，如果线程池取不到空闲的工作者线程，还需要做一些处理。

 

2.2 线程池实现要点

server启动时，创建一定数量的工作者线程加入线程池，如（20个），供I/O线程来取用；

每当I/O线程请求空闲工作者线程时，从池中取出一个空闲工作者线程，处理相应请求；

当请求处理完毕，关闭相应I/O连接时，回收相应线程并放回线程池中供下次使用；

若请求空闲工作者线程池时，没有空闲工作者线程，可作如下处理：

(1)若池中"管理"的线程总数不超过最大允许值，可创建一批新的工作者线程加入池中，并返回其中一个供I/O线程使用；

(2)若池中"管理"的线程总数已经达到最大值，不应再继续创建新线程， 则等待一小段时间并重试。注意因为I/O线程是单线程且不应被阻塞等待在此处，所以其实对线程池的管理应由一个专门的管理线程完成，包括创建新工作者线程等工作。此时管理线程阻塞等待（如使用条件变量并等待唤醒），一小段时间之后，线程池中应有空闲工作者线程可使用。否则server负荷估计是出了问题。 


-----------

epoll是linux下高并发服务器的完美方案，因为是基于事件触发的，所以比select快的不只是一个数量级。
单线程epoll，触发量可达到15000，但是加上业务后，因为大多数业务都与数据库打交道，所以就会存在阻塞的情况，这个时候就必须用多线程来提速。
 
epoll在线程池内，测试结果2000个/s
增加了网络断线后的无效socket检测。
 
测试工具：stressmark
因为加了适用与ab的代码，所以也可以适用ab进行压力测试。
char buf[1000] = {0};
sprintf(buf,"HTTP/1.0 200 OK\r\nContent-type: text/plain\r\n\r\n%s","Hello world!\n");
send(socketfd,buf, strlen(buf),0);

[cpp] view plain copy
  1. #include <stdio.h>  
  2. #include <sys/epoll.h>  
  3. #include <sys/socket.h>  
  4. #include <netinet/in.h>  
  5. #include <arpa/inet.h>  
  6. #include <netdb.h>  
  7. #include <sys/types.h>  
  8. #include <signal.h>  
  9. #include <unistd.h>  
  10. #include <fcntl.h>  
  11. #include <string.h>  
  12. #include <errno.h>  
  13. #include <stdlib.h>  
  14.   
  15.   
  16. //stl head  
  17.   
  18. #include <ext/hash_map> //包含hash_map 的头文件  
  19.   
  20. //#include //stl的map  
  21.   
  22. using namespace std; //std 命名空间  
  23.   
  24. using namespace __gnu_cxx; //而hash_map是在__gnu_cxx的命名空间里的  
  25.   
  26.   
  27.   
  28. int init_thread_pool(int threadNum);  
  29. void *epoll_loop(void* para);  
  30. void *check_connect_timeout(void* para);  
  31.   
  32.   
  33. struct sockStruct  
  34. {  
  35.     time_t time;  
  36.   
  37.     unsigned int* recvBuf;  
  38. };  
  39.   
  40. //hash-map  
  41.   
  42. //hash_map        sock_map;  
  43.   
  44. hash_map<int, sockStruct>        sock_map;  
  45.   
  46.    
  47. #define MAXRECVBUF 4096  
  48. #define MAXBUF MAXRECVBUF+10   
  49.   
  50. int fd_Setnonblocking(int fd)  
  51. {  
  52.     int op;  
  53.    
  54.     op=fcntl(fd,F_GETFL,0);  
  55.     fcntl(fd,F_SETFL,op|O_NONBLOCK);  
  56.    
  57.     return op;  
  58. }  
  59.    
  60. void on_sigint(int signal)  
  61. {  
  62.     exit(0);  
  63. }  
  64.    
  65. /*  
  66. handle_message - 处理每个 socket 上的消息收发  
  67. */   
  68. int handle_message(int new_fd)   
  69. {   
  70.     char buf[MAXBUF + 1];   
  71.     char sendbuf[MAXBUF+1];   
  72.     int len;   
  73.     /* 开始处理每个新连接上的数据收发 */   
  74.     bzero(buf, MAXBUF + 1);   
  75.     /* 接收客户端的消息 */   
  76.     //len = recv(new_fd, buf, MAXBUF, 0);  
  77.   
  78.   
  79.   
  80.     int nRecvBuf = MAXRECVBUF; //设置为32K   
  81.   
  82.     setsockopt(new_fd, SOL_SOCKET, SO_RCVBUF, ( const char* )&nRecvBuf, sizeof(int));  
  83.     len=recv(new_fd,&buf, MAXBUF,0);  
  84.   
  85.     //--------------------------------------------------------------------------------------------  
  86.   
  87.     //这块为了使用ab测试  
  88.   
  89.     char bufSend[1000] = {0};  
  90.     sprintf(bufSend,"HTTP/1.0 200 OK\r\nContent-type: text/plain\r\n\r\n%s","Hello world!\n");  
  91.     send(new_fd,bufSend,strlen(buf),0);  
  92.   
  93.     //--------------------------------------------------------------------------------------------  
  94.   
  95.   
  96.     if (len > 0){   
  97.   
  98.         //printf ("%d接收消息成功:'%s'，共%d个字节的数据\n", new_fd, buf, len);   
  99.   
  100.   
  101.         //hash-map  
  102.   
  103.           
  104.         hash_map<int, sockStruct>::iterator it_find;  
  105.         it_find = sock_map.find(new_fd);  
  106.         if(it_find == sock_map.end()){  
  107.             //新的网络连接，申请新的接收缓冲区，并放入map中  
  108.   
  109.             //printf("new socket %d\n", new_fd);  
  110.   
  111.   
  112.             sockStruct newSockStruct;  
  113.             newSockStruct.time = time((time_t*)0);  
  114.             newSockStruct.recvBuf = (unsigned int*)malloc(1000);  
  115.             memset(newSockStruct.recvBuf, 0, 1000);  
  116.             strcat((char*)newSockStruct.recvBuf, buf);  
  117.             sock_map.insert(pair<int,sockStruct>(new_fd, newSockStruct));  
  118.         }else{  
  119.             //网络连接已经存在，找到对应的数据缓冲区，将接收到的数据拼接到数据缓冲区中  
  120.   
  121.             //printf("socket %d exist!\n", it_find->first);  
  122.   
  123.   
  124.             (it_find->second).time = time((time_t*)0);                //时间更改  
  125.   
  126.             char* bufSockMap = (char*)(it_find->second).recvBuf;    //数据存储  
  127.   
  128.   
  129.             strcat(bufSockMap, buf);  
  130.             //printf("bufSockMap:%s\n", bufSockMap);  
  131.   
  132.         }  
  133.   
  134.   
  135.     }   
  136.     else {   
  137.         if (len < 0)   
  138.             printf ("消息接收失败！错误代码是%d，错误信息是'%s'\n",   
  139.             errno, strerror(errno));   
  140.         else {  
  141.             //将socket从map中移除  
  142.   
  143.             /* 
  144.             hash_map::iterator it_find; 
  145.             it_find = sock_map.find(new_fd); 
  146.             sock_map.erase(it_find); 
  147.             */  
  148.             printf("client %d quit!\n",new_fd);   
  149.         }  
  150.         //close(new_fd);   
  151.   
  152.         return -1;   
  153.     }   
  154.     /* 处理每个新连接上的数据收发结束 */   
  155.   
  156.     //关闭socket的时候，要释放接收缓冲区。  
  157.   
  158.     hash_map<int, sockStruct>::iterator it_find;  
  159.     it_find = sock_map.find(new_fd);  
  160.     free((it_find->second).recvBuf);  
  161.     sock_map.erase(it_find);  
  162.   
  163.     close(new_fd);  
  164.     return len;   
  165. }   
  166.   
  167.   
  168.     int listenfd;  
  169.     int sock_op=1;  
  170.     struct sockaddr_in address;  
  171.     struct epoll_event event;  
  172.     struct epoll_event events[1024];  
  173.     int epfd;  
  174.     int n;  
  175.     int i;  
  176.     char buf[512];  
  177.     int off;  
  178.     int result;  
  179.     char *p;  
  180.   
  181. int main(int argc,char* argv[])  
  182. {  
  183.   
  184.     init_thread_pool(1);  
  185.   
  186.     signal(SIGPIPE,SIG_IGN);  
  187.     signal(SIGCHLD,SIG_IGN);  
  188.     signal(SIGINT,&on_sigint);  
  189.     listenfd=socket(AF_INET,SOCK_STREAM,0);  
  190.     setsockopt(listenfd,SOL_SOCKET,SO_REUSEADDR,&sock_op,sizeof(sock_op));  
  191.    
  192.     memset(&address,0,sizeof(address));  
  193.     address.sin_addr.s_addr=htonl(INADDR_ANY);  
  194.     address.sin_port=htons(8006);  
  195.     bind(listenfd,(struct sockaddr*)&address,sizeof(address));  
  196.     listen(listenfd,1024);  
  197.     fd_Setnonblocking(listenfd);  
  198.    
  199.     epfd=epoll_create(65535);  
  200.     memset(&event,0,sizeof(event));  
  201.     event.data.fd=listenfd;  
  202.     event.events=EPOLLIN|EPOLLET;  
  203.     epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&event);  
  204.   
  205.     while(1){  
  206.         sleep(1000);  
  207.     }  
  208.     return 0;  
  209. }  
  210.   
  211. /************************************************* 
  212. * Function: * init_thread_pool 
  213. * Description: * 初始化线程 
  214. * Input: * threadNum:用于处理epoll的线程数 
  215. * Output: *  
  216. * Others: * 此函数为静态static函数, 
  217. *************************************************/  
  218. int init_thread_pool(int threadNum)  
  219. {  
  220.     int i,ret;  
  221.   
  222.     pthread_t threadId;  
  223.   
  224.     //初始化epoll线程池  
  225.   
  226.     for ( i = 0; i < threadNum; i++)  
  227.     {  
  228.   
  229.         ret = pthread_create(&threadId, 0, epoll_loop, (void *)0);  
  230.         if (ret != 0)  
  231.         {  
  232.             printf("pthread create failed!\n");  
  233.             return(-1);  
  234.         }  
  235.     }  
  236.   
  237.     ret = pthread_create(&threadId, 0, check_connect_timeout, (void *)0);  
  238.   
  239.     return(0);  
  240. }  
  241. /************************************************* 
  242. * Function: * epoll_loop 
  243. * Description: * epoll检测循环 
  244. * Input: *  
  245. * Output: *  
  246. * Others: *  
  247. *************************************************/  
  248. static int count111 = 0;  
  249. static time_t oldtime = 0, nowtime = 0;  
  250. void *epoll_loop(void* para)  
  251. {  
  252.         while(1)  
  253.     {  
  254.         n=epoll_wait(epfd,events,4096,-1);  
  255.         //printf("n = %d\n", n);  
  256.   
  257.         if(n>0)  
  258.         {  
  259.             for(i=0;i<n;++i)  
  260.             {  
  261.                 if(events[i].data.fd==listenfd)  
  262.                 {  
  263.                     while(1)  
  264.                     {  
  265.                         event.data.fd=accept(listenfd,NULL,NULL);  
  266.                         if(event.data.fd>0)  
  267.                         {  
  268.                             fd_Setnonblocking(event.data.fd);  
  269.                             event.events=EPOLLIN|EPOLLET;  
  270.                             epoll_ctl(epfd,EPOLL_CTL_ADD,event.data.fd,&event);  
  271.                         }  
  272.                         else  
  273.                         {  
  274.                             if(errno==EAGAIN)  
  275.                             break;  
  276.                         }  
  277.                     }  
  278.                 }  
  279.                 else  
  280.                 {  
  281.                     if(events[i].events&EPOLLIN)  
  282.                     {  
  283.                         //handle_message(events[i].data.fd);  
  284.   
  285.   
  286.                         char recvBuf[1024] = {0};   
  287.   
  288.                         int ret = 999;  
  289.   
  290.                         int rs = 1;  
  291.   
  292.   
  293.                         while(rs)  
  294.                         {  
  295.                             ret = recv(events[n].data.fd,recvBuf,1024,0);// 接受客户端消息  
  296.   
  297.                             if(ret < 0)  
  298.                             {  
  299.                                 //由于是非阻塞的模式,所以当errno为EAGAIN时,表示当前缓冲区已无数据可//读在这里就当作是该次事件已处理过。  
  300.   
  301.                                 if(errno == EAGAIN)  
  302.                                 {  
  303.                                     printf("EAGAIN\n");  
  304.                                     break;  
  305.                                 }  
  306.                                 else{  
  307.                                     printf("recv error!\n");  
  308.                                     epoll_ctl(epfd, EPOLL_CTL_DEL, events[i].data.fd, &event);  
  309.                                     close(events[i].data.fd);  
  310.                                     break;  
  311.                                 }  
  312.                             }  
  313.                             else if(ret == 0)  
  314.                             {  
  315.                                 // 这里表示对端的socket已正常关闭.   
  316.   
  317.                                 rs = 0;  
  318.                             }  
  319.                             if(ret == sizeof(recvBuf))  
  320.                                 rs = 1; // 需要再次读取  
  321.   
  322.                             else  
  323.                                 rs = 0;  
  324.                         }  
  325.   
  326.   
  327.   
  328.   
  329.                         if(ret>0){  
  330.   
  331.                             count111 ++;  
  332.   
  333.   
  334.   
  335.                             struct tm *today;  
  336.                             time_t ltime;  
  337.                             time( &nowtime );  
  338.   
  339.                             if(nowtime != oldtime){  
  340.                                 printf("%d\n", count111);  
  341.                                 oldtime = nowtime;  
  342.                                 count111 = 0;  
  343.                             }  
  344.   
  345.   
  346.                             char buf[1000] = {0};  
  347.                             sprintf(buf,"HTTP/1.0 200 OK\r\nContent-type: text/plain\r\n\r\n%s","Hello world!\n");  
  348.                             send(events[i].data.fd,buf,strlen(buf),0);  
  349.   
  350.   
  351.                             //    CGelsServer Gelsserver;  
  352.   
  353.                             //    Gelsserver.handle_message(events[i].data.fd);  
  354.   
  355.                         }  
  356.   
  357.   
  358.                         epoll_ctl(epfd, EPOLL_CTL_DEL, events[i].data.fd, &event);  
  359.                         close(events[i].data.fd);  
  360.   
  361.                     }  
  362.                     else if(events[i].events&EPOLLOUT)  
  363.                     {  
  364.                         sprintf(buf,"HTTP/1.0 200 OK\r\nContent-type: text/plain\r\n\r\n%s","Hello world!\n");  
  365.                         send(events[i].data.fd,buf,strlen(buf),0);  
  366.                         /* 
  367.                         if(p!=NULL) 
  368.                         { 
  369.                             free(p); 
  370.                             p=NULL; 
  371.                         } 
  372.                         */  
  373.                         close(events[i].data.fd);  
  374.                     }  
  375.                     else  
  376.                     {  
  377.                         close(events[i].data.fd);  
  378.                     }  
  379.                 }  
  380.             }  
  381.         }  
  382.     }  
  383.   
  384. }  
  385. /************************************************* 
  386. * Function: * check_connect_timeout 
  387. * Description: * 检测长时间没反应的网络连接，并关闭删除 
  388. * Input: *  
  389. * Output: *  
  390. * Others: *  
  391. *************************************************/  
  392. void *check_connect_timeout(void* para)  
  393. {  
  394.     hash_map<int, sockStruct>::iterator it_find;  
  395.     for(it_find = sock_map.begin(); it_find!=sock_map.end(); ++it_find){  
  396.         if( time((time_t*)0) - (it_find->second).time > 120){                //时间更改  
  397.   
  398.   
  399.             free((it_find->second).recvBuf);  
  400.             sock_map.erase(it_find);  
  401.   
  402.             close(it_find->first);  
  403.         }  
  404.     }  
  405.   
  406. }  

相关推荐