首页 诗词 字典 板报 句子 名言 友答 励志 学校 网站地图
当前位置: 首页 > 教程频道 > 开发语言 > 编程 >

epoll服务器示范, 监听5000个端口, 使用线程池

2012-09-02 
epoll服务器示例, 监听5000个端口, 使用线程池运行这个程序需要预先设置栈内存和文件描述符上限, 否则运行

epoll服务器示例, 监听5000个端口, 使用线程池

运行这个程序需要预先设置栈内存和文件描述符上限, 否则运行失败
ulimit -n 16384
ulimit -s 4096

文件名:server.c
编译: gcc server.c -Wall -O2 -pthread -o server
程序源码如下(请自行编辑宏定义SERVER_IP为自己的IP):

?

/*Linux 2.6 x86_64 only*/#include <stdio.h>#include <string.h>#include <stdlib.h>#include <time.h>#include <unistd.h>#include <sys/epoll.h>#include <sys/stat.h>#include <sys/types.h>#include <sys/socket.h>#include <netinet/in.h>#include <arpa/inet.h>#include <unistd.h>#include <netdb.h>#include <pthread.h>#define THREAD_MAX 4096#define LISTEN_MAX 5000#define SERVER_IP "192.168.1.103"typedef struct {char ip4[128];int port;int fd;} LISTEN_INFO;//服务器参数static LISTEN_INFO s_listens[LISTEN_MAX];//线程池参数static unsigned int s_thread_para[THREAD_MAX][8];//线程参数static pthread_t s_tid[THREAD_MAX];//线程IDpthread_mutex_t s_mutex[THREAD_MAX];//线程锁//私有函数static int init_thread_pool(void);static int init_listen4(char *ip4, int port, int max_link);//线程函数void * test_server4(unsigned int thread_para[]);int main(int argc, char *argv[])//客户端驱动{//临时变量int i, j, rc;int sock_listen; //监听套接字int sock_cli; //客户端连接int listen_index;int epfd;int nfds;struct epoll_event ev;struct epoll_event events[LISTEN_MAX];socklen_t addrlen; //地址信息长度struct sockaddr_in addr4; //IPv4地址结构//线程池初始化rc = init_thread_pool();if (0 != rc) exit(-1);//初始化服务监听for(i = 0; i < LISTEN_MAX; i++) {sprintf(s_listens[i].ip4, "%s", SERVER_IP);s_listens[i].port = 8000 + i;//创建监听rc = init_listen4(s_listens[i].ip4, s_listens[i].port, 64);if (0 > rc) {fprintf(stderr, "无法创建服务器监听于%s:%d\r\n", s_listens[i].ip4, s_listens[i].port);exit(-1);}s_listens[i].fd = rc;}//设置集合epfd = epoll_create(8192);for (i = 0; i < LISTEN_MAX; i++) {//加入epoll事件集合ev.events = EPOLLIN;ev.data.u32 = i;//记录listen数组下标if (epoll_ctl(epfd, EPOLL_CTL_ADD, s_listens[i].fd, &ev) < 0) {fprintf(stderr, "向epoll集合添加套接字失败(fd =%d)\r\n", rc);exit(-1);}}//服务循环for( ; ; ) {//等待epoll事件nfds = epoll_wait(epfd, events, LISTEN_MAX, -1);//处理epoll事件for(i = 0; i < nfds; i++) {//接收客户端连接listen_index = events[i].data.u32;sock_listen = s_listens[listen_index].fd;addrlen = sizeof(struct sockaddr_in);bzero(&addr4, addrlen);sock_cli = accept(sock_listen, (struct sockaddr *)&addr4, &addrlen);if(0 > sock_cli) {fprintf(stderr, "接收客户端连接失败\n");continue;}//查询空闲线程对for(j = 0; j < THREAD_MAX; j++) {if (0 == s_thread_para[j][0]) break;}if (j >= THREAD_MAX) {fprintf(stderr, "线程池已满, 连接将被放弃\r\n");shutdown(sock_cli, SHUT_RDWR);close(sock_cli);continue;}//复制有关参数s_thread_para[j][0] = 1;//设置活动标志为"活动"s_thread_para[j][1] = sock_cli;//客户端连接s_thread_para[j][2] = listen_index;//服务索引//线程解锁pthread_mutex_unlock(s_mutex + j);}//end of for(i;;)}//end of for(;;)exit(0);}static int init_thread_pool(void){int i, rc;//初始化线程池参数for(i = 0; i < THREAD_MAX; i++) {s_thread_para[i][0] = 0;//设置线程占用标志为"空闲"s_thread_para[i][7] = i;//线程池索引pthread_mutex_lock(s_mutex + i);//线程锁}//创建线程池for(i = 0; i < THREAD_MAX; i++) {rc = pthread_create(s_tid + i, 0, (void *)test_server4, (void *)(s_thread_para[i]));if (0 != rc) {fprintf(stderr, "线程创建失败\n");return(-1);}}//成功返回return(0);}static int init_listen4(char *ip4, int port, int max_link){//临时变量int sock_listen4;struct sockaddr_in addr4;unsigned int optval;struct linger optval1;//初始化数据结构bzero(&addr4, sizeof(addr4));inet_pton(AF_INET, ip4, &(addr4.sin_addr));addr4.sin_family = AF_INET;addr4.sin_port = htons(port);//创建SOCKETsock_listen4 = socket(AF_INET, SOCK_STREAM, 0);if (0 > sock_listen4) return(-1);//设置SO_REUSEADDR选项(服务器快速重起)optval = 0x1;setsockopt(sock_listen4, SOL_SOCKET, SO_REUSEADDR, &optval, 4);//设置SO_LINGER选项(防范CLOSE_WAIT挂住所有套接字)optval1.l_onoff = 1;optval1.l_linger = 60;setsockopt(sock_listen4, SOL_SOCKET, SO_LINGER, &optval1, sizeof(struct linger));if (0 > bind(sock_listen4, (struct sockaddr *)&addr4, sizeof(addr4))) {close(sock_listen4);return(-1);}if (0 > listen(sock_listen4, max_link)) {close(sock_listen4);return(-1);}return(sock_listen4);}void * test_server4(unsigned int thread_para[]){//临时变量int pool_index; //线程池索引int sock_cli; //客户端连接int listen_index; //监听索引char buff[32768]; //传输缓冲区char *p;int i, j, len;//线程脱离创建者pthread_detach(pthread_self());pool_index = thread_para[7];wait_unlock:pthread_mutex_lock(s_mutex + pool_index);//等待线程解锁//线程变量内容复制sock_cli = thread_para[1];//客户端连接listen_index = thread_para[2];//监听索引//接收请求len = recv(sock_cli, buff, 32768, MSG_NOSIGNAL);//构造响应p = buff;//HTTP头p += sprintf(p, "HTTP/1.1 200 OK\r\n");p += sprintf(p, "Content-Type: text/html\r\n");p += sprintf(p, "Connection: closed\r\n\r\n");//页面p += sprintf(p, "<html>\r\n<head>\r\n");p += sprintf(p, "<meta content="text/html; charset=UTF-8" http-equiv="Content-Type">\r\n");p += sprintf(p, "</head>\r\n");p += sprintf(p, "<body style="background-color: rgb(229, 229, 229);">\r\n");p += sprintf(p, "<center>\r\n");p += sprintf(p, "<H3>连接状态</H3>\r\n");p += sprintf(p, "<p>服务器地址 %s:%d</p>\r\n", s_listens[listen_index].ip4, s_listens[listen_index].port);j = 0;for(i = 0; i < THREAD_MAX; i++) {if (0 != s_thread_para[i][0]) j++;}p += sprintf(p, "<H3>线程池状态</H3>\r\n");p += sprintf(p, "<p>线程池总数 %d 活动线程总数 %d</p>\r\n", THREAD_MAX, j);p += sprintf(p, "</center></body></html>\r\n");len = p - buff;//发送响应send(sock_cli, buff, len, MSG_NOSIGNAL);//释放连接shutdown(sock_cli, SHUT_RDWR);close(sock_cli);//线程任务结束thread_para[0] = 0;//设置线程占用标志为"空闲"goto wait_unlock;pthread_exit(NULL);}

?

?

#include <iostream>#include <sys/socket.h>#include <sys/epoll.h>#include <netinet/in.h>#include <arpa/inet.h>#include <fcntl.h>#include <unistd.h>#include <stdio.h>#include <pthread.h> #define MAXLINE 10#define OPEN_MAX 100#define LISTENQ 20#define SERV_PORT 5555#define INFTIM 1000 //线程池任务队列结构体struct task{  int fd;            //需要读写的文件描述符  struct task *next; //下一个任务}; //用于读写两个的两个方面传递参数struct user_data{  int fd;  unsigned int n_size;  char line[MAXLINE];}; //线程的任务函数void * readtask(void *args);void * writetask(void *args);  //声明epoll_event结构体的变量,ev用于注册事件,数组用于回传要处理的事件struct epoll_event ev,events[20];int epfd;pthread_mutex_t mutex;pthread_cond_t cond1;struct task *readhead=NULL,*readtail=NULL,*writehead=NULL; void setnonblocking(int sock){     int opts;     opts=fcntl(sock,F_GETFL);     if(opts<0)     {          perror("fcntl(sock,GETFL)");          exit(1);     }    opts = opts|O_NONBLOCK;     if(fcntl(sock,F_SETFL,opts)<0)     {          perror("fcntl(sock,SETFL,opts)");          exit(1);     }   } int main(){     int i, maxi, listenfd, connfd, sockfd,nfds;     pthread_t tid1,tid2;         struct task *new_task=NULL;     struct user_data *rdata=NULL;     socklen_t clilen;         pthread_mutex_init(&mutex,NULL);     pthread_cond_init(&cond1,NULL);     //初始化用于读线程池的线程     pthread_create(&tid1,NULL,readtask,NULL);     pthread_create(&tid2,NULL,readtask,NULL);         //生成用于处理accept的epoll专用的文件描述符        epfd=epoll_create(256);      struct sockaddr_in clientaddr;     struct sockaddr_in serveraddr;     listenfd = socket(AF_INET, SOCK_STREAM, 0);     //把socket设置为非阻塞方式     setnonblocking(listenfd);     //设置与要处理的事件相关的文件描述符     ev.data.fd=listenfd;     //设置要处理的事件类型     ev.events=EPOLLIN|EPOLLET;     //注册epoll事件     epoll_ctl(epfd,EPOLL_CTL_ADD,listenfd,&ev);         bzero(&serveraddr, sizeof(serveraddr));     serveraddr.sin_family = AF_INET;         char *local_addr="200.200.200.222";     inet_aton(local_addr,&(serveraddr.sin_addr));//htons(SERV_PORT);     serveraddr.sin_port=htons(SERV_PORT);     bind(listenfd,(sockaddr *)&serveraddr, sizeof(serveraddr));     listen(listenfd, LISTENQ);         maxi = 0;     for ( ; ; ) {          //等待epoll事件的发生          nfds=epoll_wait(epfd,events,20,500);          //处理所发生的所有事件             for(i=0;i<nfds;++i)        {               if(events[i].data.fd==listenfd)               {                                       connfd = accept(listenfd,(sockaddr *)&clientaddr, &clilen);                    if(connfd<0){                      perror("connfd<0");                      exit(1);                   }                    setnonblocking(connfd);                                       char *str = inet_ntoa(clientaddr.sin_addr);                    std::cout<<"connec_ from >>"<<str<<std::endl;                    //设置用于读操作的文件描述符                    ev.data.fd=connfd;                    //设置用于注测的读操作事件                 ev.events=EPOLLIN|EPOLLET;                    //注册ev                 epoll_ctl(epfd,EPOLL_CTL_ADD,connfd,&ev);               }            else if(events[i].events&EPOLLIN)            {                    printf("reading!\n");                                    if ( (sockfd = events[i].data.fd) < 0) continue;                    new_task=new task();                    new_task->fd=sockfd;                    new_task->next=NULL;                    //添加新的读任务                    pthread_mutex_lock(&mutex);                    if(readhead==NULL)                    {                      readhead=new_task;                      readtail=new_task;                    }                       else                    {                        readtail->next=new_task;                      readtail=new_task;                    }                      //唤醒所有等待cond1条件的线程                    pthread_cond_broadcast(&cond1);                    pthread_mutex_unlock(&mutex);                }               else if(events[i].events&EPOLLOUT)               {                 rdata=(struct user_data *)events[i].data.ptr;                 sockfd = rdata->fd;                 write(sockfd, rdata->line, rdata->n_size);                 delete rdata;                 //设置用于读操作的文件描述符                 ev.data.fd=sockfd;                 //设置用于注测的读操作事件               ev.events=EPOLLIN|EPOLLET;                 //修改sockfd上要处理的事件为EPOLIN               epoll_ctl(epfd,EPOLL_CTL_MOD,sockfd,&ev);               }                                       }              }}void * readtask(void *args){      int fd=-1;   unsigned int n;   //用于把读出来的数据传递出去   struct user_data *data = NULL;   while(1){                pthread_mutex_lock(&mutex);        //等待到任务队列不为空        while(readhead==NULL)             pthread_cond_wait(&cond1,&mutex);                fd=readhead->fd;        //从任务队列取出一个读任务        struct task *tmp=readhead;        readhead = readhead->next;        delete tmp;        pthread_mutex_unlock(&mutex);        data = new user_data();        data->fd=fd;        if ( (n = read(fd, data->line, MAXLINE)) < 0) {                      if (errno == ECONNRESET) {             close(fd);                       } else             std::cout<<"readline error"<<std::endl;           if(data!=NULL)delete data;        } else if (n == 0) {            close(fd);           printf("Client close connect!\n");           if(data!=NULL)delete data;        } else{                data->n_size=n;        //设置需要传递出去的数据        ev.data.ptr=data;        //设置用于注测的写操作事件        ev.events=EPOLLOUT|EPOLLET;        //修改sockfd上要处理的事件为EPOLLOUT        epoll_ctl(epfd,EPOLL_CTL_MOD,fd,&ev);       }   }}
?

热点排行