线程池.cc

linux线程编程

#include <pthread.h> 
#include <sys/types.h> 
#include <sys/ipc.h> 
#include <sys/msg.h> 
#include <unistd.h> 
#include <iostream> 
#include <sys/socket.h> 
#include <netinet/in.h> 
#include <arpa/inet.h> 
#include <sys/uio.h> 
#include <errno.h> 
#include <signal.h> 
#include <string.h> 

using namespace std; 
#define MAXLINE 1024 
#define ERR_REPORT() { cout << strerror(errno) << endl; exit(-1); } 

typedef struct mymsg 
{ 
       long mtype; 
       char mtext[MAXLINE]; 
} _mesg; 

void sig_proc(int s) 
{ 
       return; 
} 

void* processing(void* v) 
{ 
       cout << " # " << pthread_self() << " is serving you. " << endl; 
       _mesg msg;       
       while(1) 
       { 
               memset(msg.mtext,0x00,MAXLINE); 
               if(msgrcv(*((int*)v),&msg,sizeof(msg),0,0)<0) 
               { 
                       cout << strerror(errno) << endl; 
               } 
               cout << " Data : [" << msg.mtext << "] (reported by # " << pthread_self() << ")" << endl; 
       } 
} 

class ThreadPool 
{ 
       int total; 
       int msqid; 
       void _start_listener(int port) 
       { 
               int fd = socket(AF_INET,SOCK_DGRAM,0); 
               if(fd < 0 ) ERR_REPORT(); 

               struct sockaddr_in serv,client; 
               serv.sin_family = AF_INET; 
               serv.sin_port = htons(port); 
               serv.sin_addr.s_addr = htonl(INADDR_ANY); 

               if(bind(fd,(const struct sockaddr*)&serv,sizeof(serv))<0) ERR_REPORT(); 
               cout << " UDP thread pool sample is listening on " << port << endl; 
                
               _mesg buf; 
               socklen_t len = sizeof(client); 
               while(1) 
               { 
                       memset(buf.mtext,0x00,MAXLINE); 
                       if(recvfrom(fd,buf.mtext,MAXLINE,0,(struct sockaddr*)&client,&len)<0)   
                               break; 
                       buf.mtype = 1; 
                       if(msgsnd(msqid,&buf,sizeof(buf),0)<0) ERR_REPORT(); 
               } 
       } 
public: 
       ThreadPool(int _i = 1) : total(_i) {cout << " Create thread pool " << endl; } 
       ~ThreadPool() { 
               cout << " Thread pool canceled. " << endl; 
               msgctl(msqid,IPC_RMID,0); 
        } 
       void start(int port = 5000) 
       { 
               cout << " registing signal .... Control-C ...... "; 
               if(signal(SIGINT,sig_proc)<0) ERR_REPORT(); 
               cout << " \t Done. " << endl; 
               cout << " Constructing thread pool ... ... "; 
               //create ipc queue to send/receive messages 
               msqid = msgget(0x500, IPC_CREAT | IPC_EXCL | 0600); 
               if (msqid < 0) 
               { 
                       ERR_REPORT();   
               } 
               //create threads 
               for(int i = 0; i < total; i++) 
               { 
                       pthread_t tid; 
                       if(pthread_create(&tid,NULL,processing,&msqid)<0) exit(-1); 
                       pthread_detach(tid); 
               } 
               cout << "\t Done. " << endl; 
               _start_listener(port); 
       }       
}; 

int main() 
{ 
       ThreadPool tp(10); 
       tp.start(); 
       return 0; 

} 

－－－－－－－－－－－－－供测试该服务器的客户端－－－－－－－－－ 
//Author : ZHANG Jian 
//Copyright: Tarena Inc. 
//Date : Jan 6, 2005 
//Usage : Please link the following shared object -lsocket -lnsl 

#include <iostream> 
#include <sys/types.h> 
#include <sys/socket.h> 
#include <netinet/in.h> 
#include <arpa/inet.h> 
#include <unistd.h> 
#include <cstdio> 
#include <cstdlib> 
#include <errno.h> 
#include <strings.h> 
#include <time.h> 
using namespace std; 

#define ERR_QUIT(arg) { cout << #arg << endl; exit(-1);} 

int main(int argc, char* argv[]) 
{ 
       if(argc != 3) ERR_QUIT("usage: udp_client <IP> <PORT>"); 
        
       //define socket for UDP 
       int sockfd; 
       size_t len; 
       struct sockaddr_in serv; 
       if((sockfd = socket(AF_INET, SOCK_DGRAM, 0))<0) ERR_QUIT("socket error"); 
        
       bzero(&serv,sizeof(serv)); 
       serv.sin_family = AF_INET; 
       serv.sin_port = htonl(atoi(argv[2])); 
       if(inet_pton(AF_INET,argv[1],&serv.sin_addr)<0) ERR_QUIT("inet_pton error"); 
        
       char buf[2048]; 
       int i= 0; 
       while(true) 
       { 
               if(++i >= 500) break; 
               bzero(buf,2048); 
               time_t t = time(0); 
               strcpy(buf,ctime(&t));   
               int len; 
               socklen_t c_len = sizeof(serv); 
               len = sendto(sockfd,buf,strlen(buf),0,(struct sockaddr*)&serv,c_len); 
               if(len != strlen(buf)) 
               { 
                       close(sockfd); 
                       ERR_QUIT("sendto Error"); 
               } 
               bzero(buf,2048); 
       } 
        
       close(sockfd); 
       cout << " Bye !" << endl;       
       return 0;       
}