簡 易 版 線 程 池 模 型 學 習

---

1、簡易版線程池流程

1.初始化線程池

1、初始化隊列， 隊列頭， 隊列尾初始化， 隊列能力初始化（ 隊列長度），隊列鎖 初始化線程池條件變量，給子線程賦入口函數，為線程池的子線程的線程 ID 申請空間，線程池是否啟動標志初始化為 02.啟動線程池 循環啟動子線程， 設置線程池啟動標志設置為 13.主線程啟動 listen， 接收客戶端請求 加鎖， 然后將接收到的 rs_fd 放入到 que 隊列中， 解鎖， pthread_cond_signal 等待在對應條 件上的子線程 注意：在使用隊列時，需要將相關函數封裝到一個.c文件里面，為了以后的代碼重用。

2、部分數據結構

隊列數據結構 typedef struct tag_node{int newfd;struct tag_node *pnext; }node_t,*pnode_t;typedef struct {pnode_t phead,ptail; //定義頭尾指針int capability; //可以承載最多業務的最大能力int size; //當前隊列中的任務業務數pthread_mutex_t mutex; //定義隊列鎖 }que_t,*pque_t;線程池數據結構及網絡發送數據的數據結構--小火車模型 typedef struct{ //所有的數據結構que_t que; //隊列pthread_cond_t cond; //定義線程條件變量pfunc downfile; //定義子線程入口地址函數pthread_t *pth; //線程ID的數組名，即首地址int pth_num; //創建的線程數short startflag; //線程池啟動標志 }factory,*pfac;typedef struct{int len;char buf[1000]; }train,*ptrain;

3、需要完成的部分功能函數

隊列的部分功能函數： void que_init(pque_t); void que_insert(pque_t ,pnode_t); void que_get(pque_t ,pnode_t *);線程功能函數的：void factory_init(pfac p,int pth_num,pfunc pth_handle); void factory_start(pfac p);void send_data(int); int send_n(int ,char *,int);

4、頭文件

//head.h#ifndef __HEAD_H__ #define __HEAD_H__ #include<sys/stat.h> #include <sys/stat.h> #include <errno.h> #include <fcntl.h> #include <signal.h> #include <sys/select.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <stdio.h> #include <netdb.h> #include <string.h> #include <sys/types.h> #include <unistd.h> #include <stdlib.h> #include <sys/epoll.h> #include<sys/wait.h> #include<pthread.h>#endif//work_que.h #ifndef __WORK_QUE_H__ #define __WORK_QUE_H__ #include"head.h"#define CAP 1000 typedef struct tag_node{int newfd;struct tag_node *pnext; }node_t,*pnode_t;typedef struct {pnode_t phead,ptail; //定義頭尾指針int capability; //可以承載最多業務的最大能力int size; //當前隊列中的任務業務數pthread_mutex_t mutex; //定義隊列鎖 }que_t,*pque_t;void que_init(pque_t); void que_insert(pque_t ,pnode_t); void que_get(pque_t ,pnode_t *); #endif//factory.h #ifndef __FACTORY_H__ #define __FACTORY_H__ #include"head.h" #include"work_que.h"typedef void *(*pfunc)(void *); #define FILENAME "file"typedef struct{ //所有的數據結構que_t que; //隊列pthread_cond_t cond; //定義線程條件變量pfunc downfile; //定義子線程入口地址函數pthread_t *pth; //線程ID的數組名，即首地址int pth_num; //創建的線程數short startflag; //線程池啟動標志 }factory,*pfac;typedef struct{int len;char buf[1000]; }train,*ptrain;void factory_init(pfac p,int pth_num,pfunc pth_handle); void factory_start(pfac p);void send_data(int); int send_n(int ,char *,int); #endif

5、主函數代碼

#include"factory.h"void *pth_handle(void *p) {pfac pf=(pfac)p; pque_t pq=&pf->que;pnode_t pcur; //要拿的節點，因為一會兒要free掉while(1){pthread_mutex_lock(&pq->mutex); //拿節點之前，加鎖//且首先判斷隊列為不為空if(NULL==pq->phead){//為空子線程就等待pthread_cond_wait(&pf->cond,&pq->mutex);}//否則，先拿到節點，解鎖后，進行服務，傳文件que_get(pq,&pcur); //拿節點pthread_mutex_unlock(&pq->mutex); //解鎖send_data(pcur->newfd); //發送文件free(pcur); //釋放掉節點} }int main(int argc,char **argv) {if(4!=argc){printf("please input IP PORT PTHREAD_NUM\n");return -1;}factory p;bzero(&p,sizeof(p));int pth_num=atoi(argv[3]);factory_init(&p,pth_num,pth_handle);factory_start(&p);int sfd;sfd=socket(AF_INET,SOCK_STREAM,0);if(-1==sfd){perror("socket");return -1;}struct sockaddr_in ser;memset(&ser,0,sizeof(ser));ser.sin_family=AF_INET;ser.sin_port=htons(atoi(argv[2])); //需要把它置為網絡字節序ser.sin_addr.s_addr=inet_addr(argv[1]);int ret;ret=bind(sfd,(struct sockaddr *)&ser,sizeof(ser));if(-1==ret){perror("bind");return -1;}listen(sfd,pth_num);int newfd;pnode_t pnew; //拿節點pque_t pq=&p.que; //為了后面的書寫方便//主線程接受請求，插入隊列，子線程服務請求，拿走隊列節點while(1){newfd=accept(sfd,NULL,NULL);//這里是第二個是客戶端的信息pnew=(pnode_t)calloc(1,sizeof(node_t)); //這是節點呀pnew->newfd=newfd; //將連接的描述符給申請空間的新節點pthread_mutex_lock(&pq->mutex); //插入節點前先加鎖que_insert(pq,pnew); //第二個參數已經是指針了pthread_mutex_unlock(&pq->mutex); //插入后在解鎖//由于插入新節點后，就應該通知子線程有任務到達pthread_cond_signal(&p.cond); //不要混淆呀} }

6、隊列部分函數實現work_que.c

#include"work_que.h"void que_init(pque_t que) {que->capability=CAP;pthread_mutex_init(&que->mutex,NULL); } void que_insert(pque_t pq,pnode_t pnew) {if(NULL==pq->ptail) //若隊列為空{pq->phead=pnew;pq->ptail=pnew; //頭尾指針都指向pnew}else{ //否則，進行尾插法pq->ptail->pnext=pnew;pq->ptail=pnew;}pq->size++; //操作完后，大小自動加一，表示有新請求 } void que_get(pque_t pq,pnode_t *pcur) //拿是拿指向該節點的指針 {*pcur=pq->phead; //頭刪--即從頭部取節點pq->phead=pq->phead->pnext;//拿完之后需要在判斷是否為空if(NULL==pq->phead){pq->ptail=NULL; //若為空將尾指針也置為NULL}pq->size--; }

7、線程池工廠功能函數

#include"factory.h"void factory_init(pfac p,int pth_num,pfunc pth_handle) //不要混淆呀 {que_init(&p->que);pthread_mutex_init(&p->que.mutex,NULL);pthread_cond_init(&p->cond,NULL);p->downfile=pth_handle;p->pth=(pthread_t *)calloc(pth_num,sizeof(pthread_t));p->pth_num=pth_num;p->startflag=0; }void factory_start(pfac p) {int i;if(0==p->startflag){for(i=0;i<p->pth_num;i++){pthread_create(p->pth+i,NULL,p->downfile,p);}p->startflag=1; //將線程啟動標志位置為1（即啟動）} }

8、發送文件及循環發送功能函數

#include"factory.h"void send_data(int newfd) {train t;memset(&t,0,sizeof(t));strcpy(t.buf,FILENAME);t.len=strlen(FILENAME);printf("the strcpy t.buf is %s, len =%d\n",t.buf,t.len);//先發文件名int ret;ret=send_n(newfd,(char *)&t,t.len+4);if(ret==-1){perror("send");return;}int fd=open(FILENAME,O_RDONLY);if(-1==fd){perror("open");return;}struct stat filestat;fstat(fd,&filestat);t.len=sizeof(long);memcpy(t.buf,&filestat.st_size,sizeof(filestat.st_size));send_n(newfd,(char*)&t,4+t.len);while(memset(t.buf,0,sizeof(t.buf)),(t.len=read(fd,t.buf,sizeof(t.buf)))>0){send_n(newfd,(char *)&t,t.len+4);}t.len=0;send_n(newfd,(char *)&t,4+t.len);close(newfd);close(fd); }#include"factory.h"int send_n(int sfd,char *p,int len) {int total=0,ret;while(total<len){ret=send(sfd,p+total,len-total,0);//取出每一次發送了多少個total=total+ret;}return 0; }int recv_n(int sfd,char *p,int len ) {int total=0,ret;while(total<len){ret=recv(sfd,p+total,len-total,0);total=total+ret;}return 0; }

9、客戶端程序

#include<time.h> #include <sys/stat.h> #include<errno.h> #include <fcntl.h> #include <signal.h> #include <sys/select.h> #include <sys/socket.h> #include <netinet/in.h> #include <arpa/inet.h> #include <stdio.h> #include <netdb.h> #include <string.h> #include <sys/types.h> #include <unistd.h> #include <stdlib.h> #include <sys/epoll.h> #include<sys/wait.h>typedef struct{pid_t pid;int tfds;short busy; }child,*pchild;void send_fd(int fdw,int fd); void recv_fd(int fdr,int* fd); void make_child(pchild p,int len); void child_handle(int fdr); int recv_n(int,char *,int); #include"func.h"int send_n(int sfd,char *p,int len) {int total=0,ret;while(total<len){ret=send(sfd,p+total,len-total,0);//取出每一次發送了多少個total=total+ret;}return 0; }int recv_n(int sfd,char *p,int len ) {int total=0,ret;while(total<len){ret=recv(sfd,p+total,len-total,0);total=total+ret;}return 0; }

10、客戶端主程序

#include"func.h"int main(int argc,char **argv) {if(3!=argc){printf("error argcs!\n");return -1;}int sfd;sfd=socket(AF_INET,SOCK_STREAM,0);if(-1==sfd){perror("socket");return -1;}struct sockaddr_in ser;memset(&ser,0,sizeof(ser));int ret,ret1;ser.sin_family=AF_INET;ser.sin_port=htons(atoi(argv[2]));ser.sin_addr.s_addr=inet_addr(argv[1]);ret=connect(sfd,(struct sockaddr*)&ser,sizeof(ser));int len;char buf[1000]={0};recv_n(sfd,(char *)&len,sizeof(len)); //先接受文件名recv_n(sfd,buf,len);int fd=open(buf,O_WRONLY|O_CREAT,0666);if(-1==fd){perror("open");return -1;}recv_n(sfd,(char *)&len,sizeof(len)); //在發送文件大小long file_len;recv_n(sfd,(char*)&file_len,len);long l=0;time_t now,check;time(&now);check=now;while(1){recv_n(sfd,(char *)&len,sizeof(len));if(len>0){l=l+len;memset(buf,0,sizeof(buf));recv_n(sfd,buf,len);//接受完后就要寫到襲擊者邊新建的文件里write(fd,buf,len);//接受多少就寫多少time(&now);if(now>check+1){printf("Now Data is %5.2f%s\r",100*(double)l/file_len,"%");fflush(stdout);check=now;}}else{printf("Down load successful!\n");break;}}close(fd);close(sfd);return 0; }

總結

以上是生活随笔為你收集整理的简 易 版 线 程 池 模 型 学 习的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。