简单epoll多线程服务器 ITeye技术网站

简单epoll多线程服务器 - - ITeye技术网站

简单epoll多线程服务器

学习

Epollthreadpoll.cpp代码  收藏代码
  1. /* Linux 2.6 x86_64 only*/   
  2. #include <pthread.h>   
  3. #include <string.h>   
  4. #include <stdlib.h>   
  5. #include <unistd.h>   
  6. #include <stdio.h>   
  7. #include <fcntl.h>  
  8.   
  9. #include <arpa/inet.h>   
  10. #include <sys/epoll.h>   
  11. #include <sys/errno.h>  
  12. #include <sys/socket.h>   
  13.   
  14. #define THREAD_MAX 20   
  15. #define LISTEN_MAX 20   
  16. #define SERVER_IP "127.0.0.1"   
  17.   
  18. typedef struct {  
  19.     char ip4[128];   
  20.     int port;   
  21.     int fd;   
  22. } LISTEN_INFO;   
  23.   
  24. //服务器参数   
  25. static LISTEN_INFO s_listens[LISTEN_MAX];   
  26.   
  27. //线程池参数   
  28. static unsigned int s_thread_para[THREAD_MAX][8];//线程参数   
  29. static pthread_t s_tid[THREAD_MAX];//线程ID   
  30. pthread_mutex_t s_mutex[THREAD_MAX];//线程锁   
  31.   
  32. //私有函数   
  33. static int init_thread_pool(void);//初始化数据  
  34. static int init_listen4(char *ip4, int port, int max_link); //初始化监听  
  35.   
  36. //线程函数   
  37. void * test_server4(unsigned int thread_para[]);  
  38.   
  39. //设置文件描述符为NonBlock  
  40. bool setNonBlock(int fd)  
  41. {  
  42.     int flags = fcntl(fd, F_GETFL, 0);  
  43.     flags |= O_NONBLOCK;  
  44.     if(-1 == fcntl(fd, F_SETFL, flags))  
  45.         return false;  
  46.     return true;  
  47. }  
  48.    
  49. int main(int argc, char *argv[])//客户端驱动   
  50. {   
  51.     //临时变量   
  52.     int i, j, rc;   
  53.   
  54.     int sock_listen; //监听套接字   
  55.     int sock_cli; //客户端连接   
  56.     int listen_index;   
  57.   
  58.     int epfd;   
  59.     int nfds;   
  60.     struct epoll_event ev;   
  61.     struct epoll_event events[LISTEN_MAX];  
  62.        
  63.     socklen_t addrlen; //地址信息长度   
  64.     struct sockaddr_in addr4; //IPv4地址结构   
  65.   
  66.     //线程池初始化   
  67.     rc = init_thread_pool();   
  68.     if (0 != rc) exit(-1);   
  69.   
  70.     //初始化服务监听   
  71.     for(i = 0; i < LISTEN_MAX; i++) {   
  72.         sprintf(s_listens[i].ip4, "%s", SERVER_IP);   
  73.         s_listens[i].port = 40000 + i;   
  74.         //创建监听   
  75.         rc = init_listen4(s_listens[i].ip4, s_listens[i].port, 64);   
  76.         if (0 > rc) {   
  77.             fprintf(stderr, "无法创建服务器监听于%s:%d\r\n", s_listens[i].ip4, s_listens[i].port);   
  78.             exit(-1);   
  79.         } else {  
  80.             fprintf(stdout, "已创建服务器监听于%s:%d\r\n", s_listens[i].ip4, s_listens[i].port);    
  81.         }   
  82.         s_listens[i].fd = rc;   
  83.     }   
  84.        
  85.     //设置集合   
  86.     epfd = epoll_create(8192);   
  87.     for (i = 0; i < LISTEN_MAX; i++) {   
  88.         //加入epoll事件集合   
  89.         ev.events = EPOLLIN | EPOLLET;  
  90.         ev.data.u32 = i;//记录listen数组下标   
  91.         if (epoll_ctl(epfd, EPOLL_CTL_ADD, s_listens[i].fd, &ev) < 0) {   
  92.             fprintf(stderr, "向epoll集合添加套接字失败(fd =%d)\r\n", rc);   
  93.             exit(-1);   
  94.         }   
  95.     }   
  96.        
  97.     //服务循环   
  98.     for( ; ; ) {   
  99.         //等待epoll事件   
  100.         nfds = epoll_wait(epfd, events, LISTEN_MAX, -1);   
  101.         //处理epoll事件   
  102.         for(i = 0; i < nfds; i++) {   
  103.             //接收客户端连接   
  104.             listen_index = events[i].data.u32;   
  105.             sock_listen = s_listens[listen_index].fd;   
  106.             addrlen = sizeof(struct sockaddr_in);   
  107.             bzero(&addr4, addrlen);   
  108.                
  109.             sock_cli = accept(sock_listen, (struct sockaddr *)&addr4, &addrlen);   
  110.             if(0 > sock_cli) {   
  111.                 fprintf(stderr, "接收客户端连接失败\n");   
  112.                 continue;   
  113.             } else {  
  114.                 char *myIP = inet_ntoa(addr4.sin_addr);  
  115.                 printf("accept a connection from %s...\n", myIP);   
  116.             }   
  117.                
  118.             setNonBlock(sock_cli);  
  119.             //查询空闲线程对   
  120.             for(j = 0; j < THREAD_MAX; j++) {   
  121.                 if (0 == s_thread_para[j][0]) break;   
  122.             }   
  123.             if (j >= THREAD_MAX) {   
  124.                 fprintf(stderr, "线程池已满, 连接将被放弃\r\n");   
  125.                 shutdown(sock_cli, SHUT_RDWR);   
  126.                 close(sock_cli);   
  127.                 continue;   
  128.             }   
  129.             //复制有关参数   
  130.             s_thread_para[j][0] = 1;//设置活动标志为"活动"   
  131.             s_thread_para[j][1] = sock_cli;//客户端连接   
  132.             s_thread_para[j][2] = listen_index;//服务索引   
  133.             //线程解锁   
  134.             pthread_mutex_unlock(s_mutex + j);   
  135.         }//end of for(i;;)   
  136.     }//end of for(;;)   
  137.   
  138.     exit(0);   
  139. }   
  140.   
  141. static int init_thread_pool(void)   
  142. {   
  143.     int i, rc;   
  144.   
  145.     //初始化线程池参数   
  146.     for(i = 0; i < THREAD_MAX; i++) {   
  147.         s_thread_para[i][0] = 0;//设置线程占用标志为"空闲"   
  148.         s_thread_para[i][7] = i;//线程池索引   
  149.         pthread_mutex_lock(s_mutex + i);// 这个地方为什么要加锁？不加锁创建监听有时会不成功   
  150.     }   
  151.   
  152.     //创建线程池   
  153.     for(i = 0; i < THREAD_MAX; i++) {   
  154.         rc = pthread_create(s_tid + i, 0, (void* (*)(void *))test_server4, (void *)(s_thread_para[i]));   
  155.         if (0 != rc) {   
  156.             fprintf(stderr, "线程创建失败\n");   
  157.             return(-1);   
  158.         }   
  159.     }   
  160.   
  161.     //成功返回   
  162.     return(0);   
  163. }   
  164.   
  165. static int init_listen4(char *ip4, int port, int max_link)   
  166. {   
  167.     //临时变量   
  168.     int sock_listen4;   
  169.     struct sockaddr_in addr4;   
  170.     unsigned int optval;   
  171.     struct linger optval1;   
  172.   
  173.     //初始化数据结构   
  174.     bzero(&addr4, sizeof(addr4));   
  175.     //inet_pton将点分十进制IP转换为整数  
  176.     inet_pton(AF_INET, ip4, &(addr4.sin_addr));   
  177.     addr4.sin_family = AF_INET;   
  178.     //htons将无符号short从主机字节序(x86:Big-Endian)转换为网络字节序  
  179.     addr4.sin_port = htons(port);   
  180.        
  181.     //创建流类型的SOCKET   
  182.     sock_listen4 = socket(AF_INET, SOCK_STREAM, 0);   
  183.     if (0 > sock_listen4) {  
  184.         fprintf(stderr, "创建socket异常, sock_listen4:%d\n", sock_listen4);  
  185.         perror("创建socket异常");  
  186.         return(-1);   
  187.     }  
  188.        
  189.     //设置SO_REUSEADDR选项(服务器快速重起)   
  190.     optval = 0x1;   
  191.     setsockopt(sock_listen4, SOL_SOCKET, SO_REUSEADDR, &optval, 4);   
  192.   
  193.     //设置SO_LINGER选项(防范CLOSE_WAIT挂住所有套接字)   
  194.     optval1.l_onoff = 1;   
  195.     optval1.l_linger = 60;   
  196.     setsockopt(sock_listen4, SOL_SOCKET, SO_LINGER, &optval1, sizeof(struct linger));   
  197.   
  198.     if (0 > bind(sock_listen4, (struct sockaddr *)&addr4, sizeof(addr4))) {   
  199.         fprintf(stderr, "bind socket异常, sock_listen4:%d\n", sock_listen4);  
  200.         perror("bind socket异常");  
  201.         close(sock_listen4);  
  202.         return(-1);   
  203.     }   
  204.   
  205.     if (0 > listen(sock_listen4, max_link)) {   
  206.         fprintf(stderr, "listen socket异常, sock_listen4:%d\n", sock_listen4);  
  207.         perror("listen socket异常");  
  208.         close(sock_listen4);   
  209.         return(-1);   
  210.     }   
  211.   
  212.     return (sock_listen4);   
  213. }   
  214.   
  215. void * test_server4(unsigned int thread_para[])   
  216. {   
  217.     //临时变量   
  218.     int sock_cli; //客户端连接   
  219.     int pool_index; //线程池索引   
  220.     int listen_index; //监听索引   
  221.   
  222.     char buff[32768]; //传输缓冲区   
  223.     int i, j, len;   
  224.     char *p;   
  225.   
  226.     //线程脱离创建者   
  227.     pthread_detach(pthread_self());   
  228.     pool_index = thread_para[7];   
  229.   
  230. wait_unlock:   
  231.     pthread_mutex_lock(s_mutex + pool_index);//等待线程解锁   
  232.   
  233.     //线程变量内容复制   
  234.     sock_cli = thread_para[1];//客户端连接   
  235.     listen_index = thread_para[2];//监听索引   
  236.   
  237.     //接收请求   
  238.     len = recv(sock_cli, buff, sizeof(buff), MSG_NOSIGNAL);   
  239.     printf("%s\n", buff);  
  240.        
  241.     //构造响应   
  242.     p = buff;   
  243.     //HTTP头   
  244.     p += sprintf(p, "HTTP/1.1 200 OK\r\n");   
  245.     p += sprintf(p, "Content-Type: text/html\r\n");   
  246.     p += sprintf(p, "Connection: closed\r\n\r\n");   
  247.     //页面   
  248.     p += sprintf(p, "<html>\r\n<head>\r\n");   
  249.     p += sprintf(p, "<meta content=\"text/html; charset=UTF-8\" http-equiv=\"Content-Type\">\r\n");   
  250.     p += sprintf(p, "</head>\r\n");   
  251.     p += sprintf(p, "<body style=\"background-color: rgb(229, 229, 229);\">\r\n");   
  252.   
  253.     p += sprintf(p, "<center>\r\n");   
  254.     p += sprintf(p, "<H3>连接状态</H3>\r\n");   
  255.     p += sprintf(p, "<p>服务器地址 %s:%d</p>\r\n", s_listens[listen_index].ip4, s_listens[listen_index].port);   
  256.     j = 0;   
  257.     for(i = 0; i < THREAD_MAX; i++) {   
  258.         if (0 != s_thread_para[i][0]) j++;   
  259.     }   
  260.     p += sprintf(p, "<H3>线程池状态</H3>\r\n");   
  261.     p += sprintf(p, "<p>线程池总数 %d 活动线程总数 %d</p>\r\n", THREAD_MAX, j);   
  262.     p += sprintf(p, "</center></body></html>\r\n");   
  263.     len = p - buff;   
  264.        
  265.     //发送响应   
  266.     send(sock_cli, buff, len, MSG_NOSIGNAL);   
  267.     memset(buff, 0, 32768);  
  268.        
  269.     //释放连接   
  270.     shutdown(sock_cli, SHUT_RDWR);   
  271.     close(sock_cli);   
  272.   
  273.     //线程任务结束   
  274.     thread_para[0] = 0;//设置线程占用标志为"空闲"   
  275.     goto wait_unlock;   
  276.   
  277.     pthread_exit(NULL);   
  278. }   
 g++ epollthreadpoll.cpp lpthread
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/lexus/p/2983597.html