這是我所用的線程池,主任務(wù)為復(fù)制文件.首先是功能函數(shù).c和.h文件

cp_dir.h

#ifndef __CP_DIR_H__ #define __CP_DIR_H__#include "head.h" #include "pthread_pool.h"#define MAXNAME_LEN 4096 struct cp_file {//要復(fù)制的文件char src[MAXNAME_LEN];//目標(biāo)文件char dest[MAXNAME_LEN]; }; //將dir_src 指向的目錄拷貝到dir_dest目錄下 void cp_dir( pthread_pool *pool, char *dir_src, char *dir_dest);//普通函數(shù)-->實現(xiàn)兩個文件之間的拷貝 void cp_file(void *arg);#endif

這是 cp_dir.c 文件

#include "cp_dir.h"//普通函數(shù)--->實現(xiàn)兩個文件之間的拷貝 void cp_file(void *arg) {//設(shè)置本線程為分離屬性//pthread_detach(pthread_self());//將兩個文件名解析出來struct cp_file *p = (struct cp_file *)arg;//以只讀的方式去打開源文件int fd_src = open(p->src,O_RDONLY);if(fd_src == -1){printf("%s",p->src);fflush(stdout);perror("open src file failed");goto cp_return;}//以可讀可寫打開目標(biāo)文件，如果目標(biāo)文件不存在的話則創(chuàng)建，//如果目標(biāo)文件存在的話則截短。int fd_dest = open(p->dest,O_RDWR | O_CREAT | O_TRUNC,0777);if(fd_dest == -1){perror("open dest file failed");goto cp_return; }int ret;char buf[1024];//讀取源文件的內(nèi)容寫入到目標(biāo)文件中去while(1){ret = read(fd_src,buf,sizeof(buf));if(ret == 0){break;}else if(ret > 0){int w = write(fd_dest,buf,ret);if(w != ret){perror("write failed");}}else{perror("read error");break;}}cp_return:close(fd_dest);close(fd_src);free(p); }//利用pool所指向的線程池將dir_src指向的目錄拷貝到dir_dest目錄下 void cp_dir(pthread_pool *pool,char *dir_src,char *dir_dest) {//獲取到要拷貝的目錄的目錄名(不帶路徑的)char *dirname = basename(dir_src);//dirname --> "20200907"//根據(jù)剛才獲取到的目錄名和目標(biāo)路徑合成目標(biāo)目錄(帶路徑)char newdirname[MAXNAME_LEN] = {0};sprintf(newdirname,"%s/%s",dir_dest,dirname);//newdirname -> "/mnt/hgfs/CS20201/20200907"printf("%s\n",newdirname);//先創(chuàng)建目標(biāo)目錄mkdir(newdirname,0777);dir_dest = newdirname;//打開要復(fù)制的目錄DIR *dir = opendir(dir_src);if(dir == NULL){perror("open src dir failed");return ;}struct dirent *dirp = NULL;//讀取目錄項while(dirp = readdir(dir)){if(!strcmp(dirp->d_name,".") || !strcmp(dirp->d_name,"..")){continue;}char file_src[MAXNAME_LEN] = {0};//獲取到目錄項的名字(帶路徑)sprintf(file_src,"%s/%s",dir_src,dirp->d_name);printf("file_src : %s\n",file_src);//獲取目錄項的屬性struct stat st;lstat(file_src,&st);//如果是普通文件或者是鏈接文件就直接創(chuàng)建一個線程去拷貝即可if(S_ISREG(st.st_mode) || S_ISLNK(st.st_mode)){//開辟一個線程去對兩個文件實現(xiàn)直接拷貝char file_dest[MAXNAME_LEN] = {0};sprintf(file_dest,"%s/%s",dir_dest,dirp->d_name);//此時要把源文件和目標(biāo)文件都傳遞給線程函數(shù) //所以需要把這個目標(biāo)文件和源文件都保存到一個結(jié)構(gòu)體中//再把這個結(jié)構(gòu)體傳遞給線程函數(shù)struct cp_file *cp = malloc(sizeof(*cp));strcpy(cp->src,file_src);strcpy(cp->dest,file_dest);printf("拷貝的兩個文件名為：\n");printf("src_file : %s\n",file_src);printf("des_file : %s\n",file_dest);printf("--------------------------------------\n");/*//創(chuàng)建一個線程去執(zhí)行copy的任務(wù)pthread_t tid;int ret = pthread_create(&tid,NULL,cp_file,(void*)cp);if(ret != 0){perror("pthread create failed");}//pthread_join(tid,NULL);*//*int ret = pthread_create(&tids[tid_num++],NULL,cp_file,(void*)cp);if(ret != 0){perror("pthread create failed");}*/add_task(pool,cp_file,(void *)cp);}else if(S_ISDIR(st.st_mode)){//如果是目錄的話則遞歸調(diào)用字節(jié)printf("拷貝的兩個目錄名為：\n");printf("src_dir : %s\n",file_src);printf("des_dir : %s\n",dir_dest);printf("--------------------------------------\n");cp_dir(pool,file_src,dir_dest);}else{printf("%s is UNKOWN TYPE!\n",file_src);}}closedir(dir);return ; }

?

?

線程池頭文件

pthread_pool_copy.h

#ifndef __PTHREAD_POOL_H__ #define __PTHREAD_POOL_H__#include "head.h"//線程池中線程的最大數(shù)量 #define MAX_THREADS_NUM 50//線程池中任務(wù)隊列最大的任務(wù)數(shù)量 #define MAX_TASKS 1000typedef struct pthread_pool {//線程池的實現(xiàn)按照項目的不同也有所不同，但大體應(yīng)該要有如下成員：//因為“任務(wù)隊列”是一種共享資源，所有我們需要互斥鎖//就是說我們需要一把互斥鎖去保存“任務(wù)隊列”pthread_mutex_t lock;//同時當(dāng)我們的任務(wù)隊列中沒有任務(wù)的時候，線程池內(nèi)的線程應(yīng)該要休眠//以避免系統(tǒng)資源的浪費，所有需要線程條件變量//線程條件變量用來表示“任務(wù)隊列”中是否有任務(wù)pthread_cond_t cond;//任務(wù)隊列(鏈表)，指向第一個需要執(zhí)行的任務(wù)//所有的線程都從這個任務(wù)鏈表中獲取任務(wù)struct task *task_list;//指向線程ID的數(shù)組，用來保存線程池中所有線程的IDpthread_t *tids;//線程池中正在服役的線程數(shù)--->線程的個數(shù)unsigned int active_threads;//線程池中任務(wù)隊列最大的任務(wù)數(shù)量unsigned int max_waiting_tasks;//線程池中任務(wù)隊列當(dāng)前的任務(wù)數(shù)量unsigned int cur_waiting_tasks;//表示是否退出程序bool shutdown;//bool-->true(1) false(0) }pthread_pool;//任務(wù)節(jié)點 struct task {//每一個任務(wù)結(jié)點保存一個任務(wù).//所謂任務(wù)實際上就是把一個文件或目錄從源目錄下拷貝到目標(biāo)目錄下//那么這個任務(wù)該如何去保存到任務(wù)節(jié)點上面去？//任務(wù)的完成是通過函數(shù)來實現(xiàn)的，所以我們?nèi)绻ケ４嬉粋€任務(wù)的話//只需要保存完成任務(wù)的函數(shù)(cp_file)的指針既可以了。//也就是說如果我們要去完成一個任務(wù)，就是去節(jié)點保存的地址上去//執(zhí)行一個函數(shù)就可以啦，函數(shù)只要執(zhí)行完了，任務(wù)也就完成了。//那么我們就需要定義一個函數(shù)指針。保存任務(wù)函數(shù)(cp_file)的地址void (*do_task)(void *arg);//另外，我們可能需要給任務(wù)函數(shù)傳遞參數(shù)(比如：文件名)void *arg;//下一個任務(wù)struct task *next; };/*init_pool:線程池的初始化函數(shù)初始化指定的線程池的，線程池中有thread_num個初始化線程。@pool:線程池指針，指向你要初始化的線程池@thread_num：你要初始化的線程池中一開始線程的數(shù)量@返回值:成功返回0，失敗返回-1。 */ int init_pool(phtread_pool *pool,unsigned int thread_num);/*routine:任務(wù)調(diào)配函數(shù)--->線程函數(shù)所有線程開始都回去執(zhí)行這個函數(shù)，此函數(shù)會不斷的從線程池的任務(wù)隊列中取任務(wù)然后交給線程去執(zhí)行取任務(wù)--->arg表示的是線程池的指針，在線程池中有任務(wù)隊列，任務(wù)隊列中有任務(wù)節(jié)點，每一個任務(wù)節(jié)點中都包含了函數(shù)指針和函數(shù)參數(shù)。 */ void *routine(void *arg);/*銷毀線程池：銷毀線程池前要保證所有的任務(wù)都已經(jīng)完成了 */ int destroy_pool(pthread_pool *pool);/*add_task:給任務(wù)隊列增加任務(wù)，把do_task指向的任務(wù)(函數(shù)指針)和arg指向的參數(shù)保存到一個任務(wù)節(jié)點中去，同時將任務(wù)節(jié)點添加到pool表示的線程池的任務(wù)隊列中去。@pool:指針，指向你要添加的任務(wù)的線程池@fun_task:你要添加的任務(wù)(cp_file)@fun_arg:你要執(zhí)行的任務(wù)的參數(shù)(兩個文件的文件名)@返回值:成功返回0，失敗返回-1。 */ int add_task(pthread_pool *pool,void (*fun_task)(void *arg),void *fun_arg);//往線程池中添加線程 int add_threads(pthread_pool *pool,unsigned int add_threads_num);//刪除線程池中的線程 int remove_threads(pthread_pool *pool,unsigned int remove_threads_num);#endif

線程池函數(shù)

#include "pthread_pool.h"/*init_pool:線程池的初始化函數(shù)初始化指定的線程池的，線程池中有thread_num個初始化線程。@pool:線程池指針，指向你要初始化的線程池@thread_num：你要初始化的線程池中一開始線程的數(shù)量@返回值:成功返回0，失敗返回-1。 */ int init_pool(pthread_pool *pool,unsigned int thread_num) {//初始化線程池結(jié)構(gòu)體//初始化線程互斥鎖pthread_mutex_init(&pool->lock,NULL);//初始化線程條件變量pthread_cond_init(&pool->cond,NULL);//創(chuàng)建一個任務(wù)節(jié)點,并且使用task_list去指向它//需要注意的是第一個任務(wù)節(jié)點時沒有初始化的,即沒有值的//因為對線程池進行初始化的時候,此時還沒有任務(wù)pool->task_list = (struct task *)malloc(sizeof(struct task));pool->task_list->next = NULL;//指向線程ID的數(shù)組進行初始化,此數(shù)組中保存了線程池中所有線程的ID//用來保存線程ID的數(shù)組空間最好是用malloc進行開辟而且這塊要足夠大pool->tids = (pthread_t *)malloc(sizeof(pthread_t)*MAX_THREADS_NUM);if(pool->task_list == NULL || pool->tids == NULL){perror("malloc memory failed");return -1;}//線程池中正在服役的線程數(shù)量-->總的線程個數(shù)pool->active_threads = thread_num;//線程池中任務(wù)隊列最大的任務(wù)數(shù)量pool->max_waiting_tasks = MAX_TASKS;//線程池中任務(wù)隊列當(dāng)前的任務(wù)數(shù)量pool->cur_waiting_tasks = 0;//不退出pool->shutdown = false;//創(chuàng)建thread_num個線程，同時記錄所有線程的ID，讓所有線程//一開始就執(zhí)行任務(wù)調(diào)配函數(shù)去執(zhí)行任務(wù)。int i;for(i = 0;i < thread_num;i++){if(pthread_create(&pool->tids[i],NULL,routine,(void *)pool) != 0){perror("create thread failed");return -1;}//打印調(diào)試信息printf("[%lu] create tids[%d] : [%lu] is success!\n",pthread_self(),i,pool->tids[i]);}return 0; }//清理函數(shù)--->防止某一個線程帶鎖退出 void handler(void *arg) {pthread_mutex_unlock((pthread_mutex_t *)arg); }/*routine:任務(wù)調(diào)配函數(shù)--->線程函數(shù)所有線程開始都回去執(zhí)行這個函數(shù)，此函數(shù)會不斷的從線程池的任務(wù)隊列中取任務(wù)然后交給線程去執(zhí)行取任務(wù)--->arg表示的是線程池的指針，在線程池中有任務(wù)隊列，任務(wù)隊列中有任務(wù)節(jié)點，每一個任務(wù)節(jié)點中都包含了函數(shù)指針和函數(shù)參數(shù)。 */ void *routine(void *arg) {struct task *p = NULL;//因為這個函數(shù)需要去訪問任務(wù)隊列，任務(wù)隊列是共享資源//所以需要上鎖//arg是你的線程池的指針.pthread_pool *pool = (pthread_pool *)arg;while(1){//在上鎖之前可以設(shè)置一個線程退出清理函數(shù)//當(dāng)線程因為某種意外情況導(dǎo)致線程夭折的時候自動執(zhí)行//我指定的清理函數(shù)(一般是解鎖為了防止線程帶鎖退出)//只能用在線程中pthread_cleanup_push(handler,(void *)&pool->lock);//獲取到線程互斥鎖，上鎖pthread_mutex_lock(&pool->lock);//任務(wù)隊列的情況分為很多種：//1.任務(wù)隊列沒有任務(wù)并且線程池還沒有結(jié)束的時候//此時,在沒有任務(wù)的時候(條件不滿足的情況下),需要等待新任務(wù)的到來//此處不能用if,必須要用while,因為一個線程被喚醒之后,有可能"搶不到"任務(wù)while(pool->cur_waiting_tasks == 0 && !pool->shutdown){//當(dāng)前線程陷入休眠pthread_cond_wait(&pool->cond,&pool->lock);//當(dāng)前線程被喚醒的條件：//1.當(dāng)有新任務(wù)的時候 2.當(dāng)我要銷毀線程池的時候}//2.任務(wù)隊列中沒有任務(wù)并且線程池要結(jié)束if(pool->cur_waiting_tasks == 0 && pool->shutdown){//解鎖pthread_mutex_unlock(&pool->lock);pthread_exit(NULL);}//3.任務(wù)隊列中有任務(wù),將任務(wù)節(jié)點從任務(wù)鏈表中取下來//找到任務(wù)鏈表中的第二個節(jié)點,因為第一個節(jié)點上面是空節(jié)點,沒有任務(wù)p = pool->task_list->next;//摘除結(jié)點,更新鏈表pool->task_list->next = p->next;p->next = NULL;pool->cur_waiting_tasks--;//獲取到線程互斥鎖，解鎖pthread_mutex_unlock(&pool->lock);//清理線程退出函數(shù)pthread_cleanup_pop(0);//當(dāng)條件滿足(任務(wù)隊列中有任務(wù)的)的時候，去執(zhí)行任務(wù).//執(zhí)行拷貝任務(wù)的時候，拷貝任務(wù)是不能被打斷的,所以我們要設(shè)置//線程屬性為不可被取消(干掉)pthread_setcancelstate(PTHREAD_CANCEL_DISABLE,NULL);//按任務(wù)節(jié)點中的任務(wù)去執(zhí)行任務(wù)(去執(zhí)行cp_file函數(shù))(p->do_task)(p->arg);//取消線程的不可被取消的屬性pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL);//釋放任務(wù)節(jié)點free(p);} }/*銷毀線程池：銷毀線程池前要保證所有的任務(wù)都已經(jīng)完成了 */ int destroy_pool(pthread_pool *pool) {//釋放所有的空間等待任務(wù)執(zhí)行完畢pool->shutdown = true;//線程池要結(jié)束了//喚醒所有的線程pthread_cond_broadcast(&pool->cond); //利用join函數(shù)回收每一個線程int i;for(i = 0;i < pool->active_threads;i++){//從第0個線程開始回收int r = pthread_join(pool->tids[i],NULL);if(r != 0){printf("Recover tids[%d] failed!\n",i);}else{printf("Recover tids[%d] success!\n",i); }}//銷毀線程互斥鎖pthread_mutex_destroy(&pool->lock);//銷毀線程條件變量pthread_cond_destroy(&pool->cond);//釋放任務(wù)節(jié)點if(pool->task_list){free(pool->task_list);//只需要釋放第一個節(jié)點即可}free(pool->tids);free(pool);return 0; }/*add_task:給任務(wù)隊列增加任務(wù)，把do_task指向的任務(wù)(函數(shù)指針)和arg指向的參數(shù)保存到一個任務(wù)節(jié)點中去，同時將任務(wù)節(jié)點添加到pool表示的線程池的任務(wù)隊列中去。@pool:指針，指向你要添加的任務(wù)的線程池@fun_task:你要添加的任務(wù)(cp_file)@fun_arg:你要執(zhí)行的任務(wù)的參數(shù)(兩個文件的文件名)@返回值:成功返回0，失敗返回-1。 */ int add_task(pthread_pool *pool,void (*fun_task)(void *arg),void *fun_arg) {//在往任務(wù)隊列中添加任務(wù)的時候，需要注意解鎖和上鎖//加入任務(wù)后要喚醒等待的線程//創(chuàng)建一個新的節(jié)點去保存任務(wù)struct task *new_task = malloc(sizeof(*new_task));new_task->next = NULL;new_task->do_task = fun_task;new_task->arg = fun_arg;//把任務(wù)節(jié)點添加到任務(wù)鏈表中去pthread_mutex_lock(&pool->lock);//任務(wù)隊列中的任務(wù)數(shù)量已達(dá)上限if(pool->cur_waiting_tasks >= MAX_TASKS){pthread_mutex_unlock(&pool->lock);printf("Too many task,add task failed!\n");//釋放任務(wù)節(jié)點空間free(new_task->arg);free(new_task);return -1;}//任務(wù)節(jié)點采用尾插法插入任務(wù)鏈表(此處不能頭插)struct task *find = pool->task_list;while(find->next){find = find->next;}//已經(jīng)找到尾結(jié)點直接添加任務(wù)到鏈表中即可find->next = new_task;pool->cur_waiting_tasks++;//訪問完任務(wù)鏈表完之后解鎖pthread_mutex_unlock(&pool->lock);//添加完之后,喚醒線程pthread_cond_signal(&pool->cond);return 0; }//往線程池中添加線程 int add_threads(pthread_pool *pool,unsigned int add_threads_num) {}//刪除線程池中的線程 int remove_threads(pthread_pool *pool,unsigned int remove_threads_num) {}

主函數(shù)頭文件

#ifndef __HEAD_H__ #define __HEAD_H__#include <stdio.h> #include <stdlib.h> #include <string.h> #include <pthread.h> #include <sys/types.h> #include <unistd.h> #include <sys/stat.h> #include <fcntl.h> #include <dirent.h> #include <libgen.h>#endif

主函數(shù)

#include "cp_dir.h"//./a.out dir_src dir_dest int main(int argc,char *argv[]) {if(argc != 3){printf("Usage : %s <dir_src> <dir_dest>!\n",argv[0]);return -1;}//創(chuàng)建一個線程池pthread_pool *pool = malloc(sizeof(*pool));//初始化線程池int ret = init_pool(pool);if(ret == -1){printf("init pthread pool failed!\n");return -1;}//利用線程池實現(xiàn)目錄之間的拷貝cp_dir(pool,argv[1],argv[2]);destroy_pool(pool);return 0; }

?

總結(jié)

以上是生活随笔為你收集整理的martin_pthread_pool的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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