聊聊Linux C下線程池的使用

作者：Jasonangel 2021-02-01 08:28:24

線程池也是多線程的處理方式。是將“生產(chǎn)者”線程提出任務添加到“任務隊列”，然后一些線程自動完成“任務隊列”上的任務。

多線程編程，創(chuàng)建一個線程，指定去完成某一個任務，等待線程的退出。雖然能夠滿足編程需求，但是當我們需要創(chuàng)建大量的線程的時候，在創(chuàng)建過程以及銷毀線程的過程中可能會消耗大量的CPU.增加很大開銷。如：文件夾的copy、WEB服務器的響應。

線程池就是用來解決類似于這樣的一個問題的，可以降低頻繁地創(chuàng)建和銷毀線程所帶來地開銷。

線程池技術(shù)思路：一般采用預創(chuàng)建線程技術(shù)，也就是提前把需要用線程先創(chuàng)建一定數(shù)目。這些線程提前創(chuàng)建好了之后，“任務隊列”里面假設沒有任務，那么就讓這些線程休眠，一旦有任務，就喚醒線程去執(zhí)行任務，任務執(zhí)行完了，也不需要去銷毀線程，直到當你想退出或者是關機時，這個時候，那么你調(diào)用銷毀線程池地函數(shù)去銷毀線程。

線程完成任務之后不會銷毀，而是自動地執(zhí)行下一個任務。而且，當任務有很多，你可以有函數(shù)接口去增加線程數(shù)量，當任務較少時，你可以有函數(shù)接口去銷毀部分線程。

如果，創(chuàng)建和銷毀線程的時間對比執(zhí)行任務的時間可以忽略不計，那么我們在這種情況下面也就沒有必要用線程池。

“任務隊列”是一個共享資源“互斥訪問”

線程池本質(zhì)上也是一個數(shù)據(jù)結(jié)構(gòu)，需要一個結(jié)構(gòu)體去描述它：

struct pthread_pool //線程池的實現(xiàn) 
{ 
 //一般會有如下成員 
 
 //互斥鎖，用來保護這個“任務隊列” 
 pthread_mutex_t lock; //互斥鎖  
  
 //線程條件變量 表示“任務隊列”是否有任務 
 pthread_cond_t cond; //條件變量 
  
 bool shutdown; //表示是否退出程序 bool：類型 false / true 
 
 //任務隊列(鏈表)，指向第一個需要指向的任務 
 //所有的線程都從任務鏈表中獲取任務 "共享資源" 
 struct task * task_list; 
  
 //線程池中有多個線程，每一個線程都有tid, 需要一個數(shù)組去保存tid 
 pthread_t * tids; //malloc()  
  
 //線程池中正在服役的線程數(shù)，當前線程個數(shù) 
 unsigned int active_threads; 
  
 //線程池任務隊列最大的任務數(shù)量 
 unsigned int max_waiting_tasks; 
  
 //線程池任務隊列上當前有多少個任務 
 unsigned int cur_waiting_tasks; 
  
 //...... 
 
}; 
 
//任務隊列(鏈表)上面的任務結(jié)點，只要能夠描述好一個任務就可以了， 
//線程會不斷地任務隊列取任務 
struct task  //任務結(jié)點  
{ 
 // 1. 任務結(jié)點表示的任務，“函數(shù)指針”指向任務要執(zhí)行的函數(shù)(cp_file) 
 void*(* do_task)(void * arg); 
  
 //2. 指針，指向任務指向函數(shù)的參數(shù)(文件描述符) 
 void * arg; 
  
 //3. 任務結(jié)點類型的指針，指向下一個任務 
 struct task * next; 
};

線程池框架代碼如下，功能自填：

操作線程池所需要的函數(shù)接口：pthread_pool.c 、pthread_pool.h

把“線程池”想象成一個外包公司，你需要去完成的就是操作線程池所提供的函數(shù)接口。

pthread_pool.c

#include "pthread_pool.h" 
 
/* 
 init_pool: 線程池初始化函數(shù)，初始化指定的線程池中有thread_num個初始線程 
 @pool：指針，指向您要初始化的那個線程池 
 @threa_num: 您要初始化的線程池中開始的線程數(shù)量 
 返回值：  
  成功 0 
  失敗 -1 
*/ 
 
int init_pool(pthread_pool * pool , unsigned int threa_num) 
{ 
 //初始化線程池的結(jié)構(gòu)體 
  
 //初始化線程互斥鎖 
 pthread_mutex_init(&pool->lock, NULL); 
  
 //初始化線程條件變量 
 pthread_cond_init(&pool->cond, NULL); 
 
 pool->shutdown = false ;// 不退出 
 
 pool->task_list = (struct task*)malloc(sizeof(struct task)); 
 
 pool->tids = (pthread_t *)malloc(sizeof(pthread_t) * MAX_ACTIVE_THREADS); 
 if(pool->task_list == NULL || pool->tids == NULL) 
 { 
  perror("malloc memery error"); 
  return -1; 
 } 
 
 pool->task_list->next = NULL; 
 
 //線程池中一開始初始化多少個線程來服役 
 pool->active_threads = threa_num; 
 
 //表示線程池中最多有多少個任務 
 pool->max_waiting_tasks = MAX_WAITING_TASKS; 
 
 //線程池中任務隊列當前的任務數(shù)量 
 pool->cur_waiting_tasks = 0; 
 
 //創(chuàng)建thread_num個線程，并且讓線程去執(zhí)行任務調(diào)配函數(shù)， 
 //記錄所有線程的tid 
 int i = 0; 
 for(i = 0; i < threa_num; i++) 
 { 
  int ret = pthread_create(&(pool->tids)[i], NULL, routine, (void*)pool); 
  if(ret != 0) 
  { 
   perror("create thread error"); 
   return -1; 
  } 
 
  printf("[%lu]:[%s] ===> tids[%d]:[%lu]",pthread_self(), 
   __FUNCTION__, i , pool->tids[i]); 
 } 
 
 return 0; 
} 
 
/* 
 routine: 任務調(diào)配函數(shù)。 
  所有線程開始都執(zhí)行此函數(shù)，此函數(shù)會不斷的從線程池的任務隊列 
  中取下任務結(jié)點，去執(zhí)行。 
   
  任務結(jié)點中包含“函數(shù)指針” h "函數(shù)參數(shù)" 
*/ 
 
void * routine(void * arg) 
{ 
 //arg表示你的線程池的指針 
  
 while() 
 { 
  //獲取線程互斥鎖，lock  
   
  //當線程池沒有結(jié)束的時候，不斷地從線程池的任務隊列取下結(jié)點 
  //去執(zhí)行。 
   
  //釋放線程互斥鎖，unlock 
   
  //釋放任務結(jié)點 
 } 
} 
 
/* 
 destroy_pool: 銷毀線程池，銷毀前要保證所有的任務已經(jīng)完成 
*/ 
 
int destroy_pool(pthread_pool * pool) 
{ 
 //釋放所有空間 等待任務執(zhí)行完畢(join)。 
 //喚醒所有線程 
 //利用join函數(shù)回收每一個線程資源。 
} 
 
/* 
 add_task：給任務隊列增加任務， 把do_task指向的任務(函數(shù)指針)和 
  arg指向的參數(shù)保存到一個任務結(jié)點，添加到pool任務隊列中。 
   
 @pool : 您要添加任務的線程池 
 @do_task ： 您需要添加的任務(cp_file) 
 @arg： 您要執(zhí)行的任務的參數(shù)(文件描述符) 
*/ 
 
int add_task(pthread_pool *pool,void*(* do_task)(void * arg), void*arg) 
{ 
 //把第二個參數(shù)和第三個參數(shù)封裝成struct task  
  
 //再把它添加到 pool->task 任務隊列中去 
  
 //注意任務隊列是一個共享資源 
  
 //假如任務后要喚醒等待的線程。 
} 
 
//如果任務多的時候，往線程池中添加線程  pthread_create 
int add_threads(pthread_pool * pool, unsigned int num); 
{ 
 //新創(chuàng)建num個線程，讓每一個線程去執(zhí)行線程調(diào)配函數(shù) 
  
 //將每一個新創(chuàng)建的線程tid，添加到pool-> tids  
} 
 
//如果任務少的時候，減少線程池中線程的數(shù)量 pthread_cancel join 
int remove_threads(pthread_pool * pool, unsigned int num) 
{ 
 //用pthread_cancel取消num個線程  
 //利用pthread_join函數(shù)去回收資源。 
}

pthread_pool.h

#ifndef __PTHREAD_POOL_H__ 
#define __PTHREAD_POOL_H__ 
 
//表示線程池中最多有多少個線程 
#define MAX_ACTIVE_THREADS 20 
 
//表示線程池中最多有多少個任務 
#define MAX_WAITING_TASKS 1024 
 
//任務隊列(鏈表)上面的任務結(jié)點，只要能夠描述好一個任務就可以了， 
//線程會不斷地任務隊列取任務 
struct task  //任務結(jié)點  
{ 
 // 1. 任務結(jié)點表示的任務，“函數(shù)指針”指向任務要執(zhí)行的函數(shù)(cp_file) 
 void*(* do_task)(void * arg); 
  
 //2. 指針，指向任務指向函數(shù)的參數(shù)(文件描述符) 
 void * arg; 
  
 //3. 任務結(jié)點類型的指針，指向下一個任務 
 struct task * next; 
}; 
 
struct pthread_pool //線程池的實現(xiàn) 
{ 
 //一般會有如下成員 
 
 //互斥鎖，用來保護這個“任務隊列” 
 pthread_mutex_t lock; //互斥鎖  
  
 //線程條件變量 表示“任務隊列”是否有任務 
 pthread_cond_t cond; //條件變量 
  
 bool shutdown; //表示是否退出程序 bool：類型 false / true 
 
 //任務隊列(鏈表)，指向第一個需要指向的任務 
 //所有的線程都從任務鏈表中獲取任務 "共享資源" 
 struct task * task_list; 
  
 //線程池中有多個線程，每一個線程都有tid, 需要一個數(shù)組去保存tid 
 pthread_t * tids; //malloc()  
  
 //線程池中正在服役的線程數(shù)，當前線程個數(shù) 
 unsigned int active_threads; 
  
 //線程池任務隊列最大的任務數(shù)量 
 unsigned int max_waiting_tasks; 
  
 //線程池任務隊列上當前有多少個任務 
 unsigned int cur_waiting_tasks; 
  
 //...... 
 
}; 
 
/* 
 init_pool: 線程池初始化函數(shù)，初始化指定的線程池中有thread_num 
  個初始線程 
 @pool：指針，指向您要初始化的那個線程池 
 @threa_num: 您要初始化的線程池中開始的線程數(shù)量 
 返回值：  
  成功 0 
  失敗 -1 
*/ 
 
int init_pool(pthread_pool * pool , unsigned int threa_num); 
 
/* 
 routine: 任務調(diào)配函數(shù)。 
  所有線程開始都執(zhí)行此函數(shù)，此函數(shù)會不斷的從線程池的任務隊列 
  中取下任務結(jié)點，去執(zhí)行。 
   
  任務結(jié)點中包含“函數(shù)指針” h "函數(shù)參數(shù)" 
*/ 
 
void * routine(void * arg); 
 
/* 
 destroy_pool: 銷毀線程池，銷毀前要保證所有的任務已經(jīng)完成 
*/ 
 
int destroy_pool(pthread_pool * pool); 
 
/* 
 add_task：給任務隊列增加任務， 把do_task指向的任務(函數(shù)指針)和 
  arg指向的參數(shù)保存到一個任務結(jié)點，添加到pool任務隊列中。 
   
 @pool : 您要添加任務的線程池 
 @do_task ： 您需要添加的任務(cp_file) 
 @arg： 您要執(zhí)行的任務的參數(shù)(文件描述符) 
*/ 
 
int add_task(pthread_pool *pool,void*(* do_task)(void * arg), void*arg); 
 
//如果任務多的時候，往線程池中添加線程  pthread_create 
int add_threads(pthread_pool * pool, unsigned int num); 
 
 
//如果任務少的時候，減少線程池中線程的數(shù)量 pthread_cancel join 
int remove_threads(pthread_pool * pool, unsigned int num); 
 
#endif

責任編輯：姜華來源：嵌入式Linux系統(tǒng)開發(fā)