簡(jiǎn) 易 版 線 程 池 模 型 學(xué) 習(xí)

---

1、簡(jiǎn)易版線程池流程

1.初始化線程池

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

2、部分?jǐn)?shù)據(jù)結(jié)構(gòu)

隊(duì)列數(shù)據(jù)結(jié)構(gòu) typedef struct tag_node{int newfd;struct tag_node *pnext; }node_t,*pnode_t;typedef struct {pnode_t phead,ptail; //定義頭尾指針int capability; //可以承載最多業(yè)務(wù)的最大能力int size; //當(dāng)前隊(duì)列中的任務(wù)業(yè)務(wù)數(shù)pthread_mutex_t mutex; //定義隊(duì)列鎖 }que_t,*pque_t;線程池?cái)?shù)據(jù)結(jié)構(gòu)及網(wǎng)絡(luò)發(fā)送數(shù)據(jù)的數(shù)據(jù)結(jié)構(gòu)--小火車模型 typedef struct{ //所有的數(shù)據(jù)結(jié)構(gòu)que_t que; //隊(duì)列pthread_cond_t cond; //定義線程條件變量pfunc downfile; //定義子線程入口地址函數(shù)pthread_t *pth; //線程ID的數(shù)組名，即首地址int pth_num; //創(chuàng)建的線程數(shù)short startflag; //線程池啟動(dòng)標(biāo)志 }factory,*pfac;typedef struct{int len;char buf[1000]; }train,*ptrain;

3、需要完成的部分功能函數(shù)

隊(duì)列的部分功能函數(shù)： void que_init(pque_t); void que_insert(pque_t ,pnode_t); void que_get(pque_t ,pnode_t *);線程功能函數(shù)的：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; //可以承載最多業(yè)務(wù)的最大能力int size; //當(dāng)前隊(duì)列中的任務(wù)業(yè)務(wù)數(shù)pthread_mutex_t mutex; //定義隊(duì)列鎖 }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{ //所有的數(shù)據(jù)結(jié)構(gòu)que_t que; //隊(duì)列pthread_cond_t cond; //定義線程條件變量pfunc downfile; //定義子線程入口地址函數(shù)pthread_t *pth; //線程ID的數(shù)組名，即首地址int pth_num; //創(chuàng)建的線程數(shù)short startflag; //線程池啟動(dòng)標(biāo)志 }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、主函數(shù)代碼

#include"factory.h"void *pth_handle(void *p) {pfac pf=(pfac)p; pque_t pq=&pf->que;pnode_t pcur; //要拿的節(jié)點(diǎn)，因?yàn)橐粫?huì)兒要free掉while(1){pthread_mutex_lock(&pq->mutex); //拿節(jié)點(diǎn)之前，加鎖//且首先判斷隊(duì)列為不為空if(NULL==pq->phead){//為空子線程就等待pthread_cond_wait(&pf->cond,&pq->mutex);}//否則，先拿到節(jié)點(diǎn)，解鎖后，進(jìn)行服務(wù)，傳文件que_get(pq,&pcur); //拿節(jié)點(diǎn)pthread_mutex_unlock(&pq->mutex); //解鎖send_data(pcur->newfd); //發(fā)送文件free(pcur); //釋放掉節(jié)點(diǎn)} }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])); //需要把它置為網(wǎng)絡(luò)字節(jié)序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; //拿節(jié)點(diǎn)pque_t pq=&p.que; //為了后面的書寫方便//主線程接受請(qǐng)求，插入隊(duì)列，子線程服務(wù)請(qǐng)求，拿走隊(duì)列節(jié)點(diǎn)while(1){newfd=accept(sfd,NULL,NULL);//這里是第二個(gè)是客戶端的信息pnew=(pnode_t)calloc(1,sizeof(node_t)); //這是節(jié)點(diǎn)呀pnew->newfd=newfd; //將連接的描述符給申請(qǐng)空間的新節(jié)點(diǎn)pthread_mutex_lock(&pq->mutex); //插入節(jié)點(diǎn)前先加鎖que_insert(pq,pnew); //第二個(gè)參數(shù)已經(jīng)是指針了pthread_mutex_unlock(&pq->mutex); //插入后在解鎖//由于插入新節(jié)點(diǎn)后，就應(yīng)該通知子線程有任務(wù)到達(dá)pthread_cond_signal(&p.cond); //不要混淆呀} }

6、隊(duì)列部分函數(shù)實(shí)現(xiàn)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) //若隊(duì)列為空{pq->phead=pnew;pq->ptail=pnew; //頭尾指針都指向pnew}else{ //否則，進(jìn)行尾插法pq->ptail->pnext=pnew;pq->ptail=pnew;}pq->size++; //操作完后，大小自動(dòng)加一，表示有新請(qǐng)求 } void que_get(pque_t pq,pnode_t *pcur) //拿是拿指向該節(jié)點(diǎn)的指針 {*pcur=pq->phead; //頭刪--即從頭部取節(jié)點(diǎn)pq->phead=pq->phead->pnext;//拿完之后需要在判斷是否為空if(NULL==pq->phead){pq->ptail=NULL; //若為空將尾指針也置為NULL}pq->size--; }

7、線程池工廠功能函數(shù)

#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; //將線程啟動(dòng)標(biāo)志位置為1（即啟動(dòng)）} }

8、發(fā)送文件及循環(huán)發(fā)送功能函數(shù)

#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);//先發(fā)文件名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);//取出每一次發(fā)送了多少個(gè)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);//取出每一次發(fā)送了多少個(gè)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)); //在發(fā)送文件大小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; }

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。