#include <stdio.h>#include <pthread.h>#include <stdlib.h>#include <signal.h> /*線程的任務(wù)隊(duì)列由,函數(shù)和參數(shù)組成,任務(wù)由鏈表來(lái)進(jìn)行管理*/typedef struct thread_worker_s{  void *(*process)(void *arg); //處理函數(shù)  void *arg;          //參數(shù)  struct thread_worker_s *next;}thread_worker_t; #define bool int#define true 1#define false 0 /*線程池中各線程狀態(tài)描述*/#define THREAD_STATE_RUN        0#define THREAD_STATE_TASK_WAITING   1#define THREAD_STATE_TASK_PROCESSING  2#define THREAD_STATE_TASK_FINISHED   3#define THREAD_STATE_EXIT       4     typedef struct thread_info_s{  pthread_t id;  int    state;   struct thread_info_s *next;}thread_info_t; static char* thread_state_map[] ={"創(chuàng)建","等待任務(wù)","處理中","處理完成","已退出"};/*線程壓縮的時(shí)候只有 0,1,2,4 狀態(tài)的線程可以銷(xiāo)毀*/  /*線程池管理器*/#define THREAD_BUSY_PERCENT 0.5  /*線程:任務(wù) = 1:2 值越小，說(shuō)明任務(wù)多,增加線程*/#define THREAD_IDLE_PERCENT 2   /*線程:任務(wù) = 2:1 值大于1,線程多于任務(wù),銷(xiāo)毀部分線程*/ typedef struct thread_pool_s{  pthread_mutex_t queue_lock ; //隊(duì)列互斥鎖,即涉及到隊(duì)列修改時(shí)需要加鎖  pthread_cond_t queue_ready; //隊(duì)列條件鎖,隊(duì)列滿足某個(gè)條件，觸發(fā)等待這個(gè)條件的線程繼續(xù)執(zhí)行,比如說(shuō)隊(duì)列滿了,隊(duì)列空了   thread_worker_t *head   ;    //任務(wù)隊(duì)列頭指針  bool    is_destroy   ;    //線程池是否已經(jīng)銷(xiāo)毀  int num;              //線程的個(gè)數(shù)  int rnum;         ;    //正在跑的線程  int knum;         ;    //已殺死的線程  int queue_size       ;    //工作隊(duì)列的大小   thread_info_t *threads   ;    //線程組id,通過(guò)pthread_join(thread_ids[0],NULL) 來(lái)執(zhí)行線程  pthread_t   display   ;    //打印線程  pthread_t   destroy   ;    //定期銷(xiāo)毀線程的線程id  pthread_t   extend    ;  float percent       ;    //線程個(gè)數(shù)于任務(wù)的比例 rnum/queue_size  int  init_num      ;  pthread_cond_t  extend_ready     ;    //如果要增加線程}thread_pool_t; /*-------------------------函數(shù)聲明----------------------*//** * 1.初始化互斥變量 * 2.初始化等待變量 * 3.創(chuàng)建指定個(gè)數(shù)的線程線程 */thread_pool_t* thread_pool_create(int num);void *thread_excute_route(void *arg);  /*調(diào)試函數(shù)*/void debug(char *message,int flag){  if(flag)    printf("%s/n",message);} void *display_thread(void *arg);/** * 添加任務(wù)包括以下幾個(gè)操作 * 1.將任務(wù)添加到隊(duì)列末尾 * 2.通知等待進(jìn)程來(lái)處理這個(gè)任務(wù) pthread_cond_singal();*/int thread_pool_add_worker(thread_pool_t *pool,void*(*process)(void *arg),void *arg); //網(wǎng)線程池的隊(duì)列中增加一個(gè)需要執(zhí)行的函數(shù)，也就是任務(wù) /** * 銷(xiāo)毀線程池,包括以下幾個(gè)部分 * 1.通知所有等待的進(jìn)程 pthread_cond_broadcase * 2.等待所有的線程執(zhí)行完 * 3.銷(xiāo)毀任務(wù)列表 * 4.釋放鎖，釋放條件 * 4.銷(xiāo)毀線程池對(duì)象 */   void *thread_pool_is_need_recovery(void *arg);void *thread_pool_is_need_extend(void *arg);void thread_pool_destory(thread_pool_t *pool);  thread_pool_t *thread_pool_create(int num){  if(num<1){    return NULL;  }  thread_pool_t *p;  p = (thread_pool_t*)malloc(sizeof(struct thread_pool_s));  if(p==NULL)    return NULL;  p->init_num = num;  /*初始化互斥變量與條件變量*/  pthread_mutex_init(&(p->queue_lock),NULL);  pthread_cond_init(&(p->queue_ready),NULL);   /*設(shè)置線程個(gè)數(shù)*/  p->num  = num;   p->rnum = num;  p->knum = 0;   p->head = NULL;  p->queue_size =0;   p->is_destroy = false;      int i=0;  thread_info_t *tmp=NULL;  for(i=0;i<num;i++){    /*創(chuàng)建線程*/    tmp= (struct thread_info_s*)malloc(sizeof(struct thread_info_s));    if(tmp==NULL){      free(p);      return NULL;    }else{      tmp->next = p->threads;      p->threads = tmp;    }    pthread_create(&(tmp->id),NULL,thread_excute_route,p);    tmp->state = THREAD_STATE_RUN;  }   /*顯示*/  pthread_create(&(p->display),NULL,display_thread,p);  /*檢測(cè)是否需要?jiǎng)討B(tài)線程*/  //pthread_create(&(p->extend),NULL,thread_pool_is_need_extend,p);  /*動(dòng)態(tài)銷(xiāo)毀*/  pthread_create(&(p->destroy),NULL,thread_pool_is_need_recovery,p);  return p;} int thread_pool_add_worker(thread_pool_t *pool,void*(*process)(void*arg),void*arg){  thread_pool_t *p= pool;  thread_worker_t *worker=NULL,*member=NULL;  worker = (thread_worker_t*)malloc(sizeof(struct thread_worker_s));  int incr=0;  if(worker==NULL){    return -1;  }  worker->process = process;  worker->arg   = arg;  worker->next  = NULL;  thread_pool_is_need_extend(pool);  pthread_mutex_lock(&(p->queue_lock));  member = p->head;  if(member!=NULL){    while(member->next!=NULL)      member = member->next;    member->next = worker;  }else{    p->head = worker;  }  p->queue_size ++;  pthread_mutex_unlock(&(p->queue_lock));  pthread_cond_signal(&(p->queue_ready));  return 1;}  void thread_pool_wait(thread_pool_t *pool){  thread_info_t *thread;  int i=0;  for(i=0;i<pool->num;i++){    thread = (thread_info_t*)(pool->threads+i);    thread->state = THREAD_STATE_EXIT;    pthread_join(thread->id,NULL);  }}void thread_pool_destory(thread_pool_t *pool){  thread_pool_t  *p   = pool;  thread_worker_t *member = NULL;   if(p->is_destroy)    return ;  p->is_destroy = true;  pthread_cond_broadcast(&(p->queue_ready));  thread_pool_wait(pool);  free(p->threads);  p->threads = NULL;  /*銷(xiāo)毀任務(wù)列表*/  while(p->head){    member = p->head;    p->head = member->next;    free(member);  }  /*銷(xiāo)毀線程列表*/  thread_info_t *tmp=NULL;  while(p->threads){    tmp = p->threads;    p->threads = tmp->next;    free(tmp);  }   pthread_mutex_destroy(&(p->queue_lock));  pthread_cond_destroy(&(p->queue_ready));  return ;}/*通過(guò)線程id,找到對(duì)應(yīng)的線程*/thread_info_t *get_thread_by_id(thread_pool_t *pool,pthread_t id){  thread_info_t *thread=NULL;  thread_info_t *p=pool->threads;  while(p!=NULL){    if(p->id==id)      return p;    p = p->next;  }  return NULL;}  /*每個(gè)線程入口函數(shù)*/void *thread_excute_route(void *arg){  thread_worker_t *worker = NULL;  thread_info_t  *thread = NULL;   thread_pool_t*  p = (thread_pool_t*)arg;  //printf("thread %lld create success/n",pthread_self());  while(1){    pthread_mutex_lock(&(p->queue_lock));     /*獲取當(dāng)前線程的id*/    pthread_t pthread_id = pthread_self();    /*設(shè)置當(dāng)前狀態(tài)*/    thread = get_thread_by_id(p,pthread_id);     /*線程池被銷(xiāo)毀,并且沒(méi)有任務(wù)了*/    if(p->is_destroy==true && p->queue_size ==0){      pthread_mutex_unlock(&(p->queue_lock));      thread->state = THREAD_STATE_EXIT;      p->knum ++;      p->rnum --;      pthread_exit(NULL);    }    if(thread){      thread->state = THREAD_STATE_TASK_WAITING; /*線程正在等待任務(wù)*/    }    /*線程池沒(méi)有被銷(xiāo)毀,沒(méi)有任務(wù)到來(lái)就一直等待*/    while(p->queue_size==0 && !p->is_destroy){      pthread_cond_wait(&(p->queue_ready),&(p->queue_lock));    }    p->queue_size--;    worker = p->head;    p->head = worker->next;    pthread_mutex_unlock(&(p->queue_lock));          if(thread)      thread->state = THREAD_STATE_TASK_PROCESSING; /*線程正在執(zhí)行任務(wù)*/    (*(worker->process))(worker->arg);    if(thread)      thread->state = THREAD_STATE_TASK_FINISHED;  /*任務(wù)執(zhí)行完成*/    free(worker);    worker = NULL;  }}   /*拓展線程*/void *thread_pool_is_need_extend(void *arg){  thread_pool_t *p = (thread_pool_t *)arg;  thread_pool_t *pool = p;  /*判斷是否需要增加線程,最終目的 線程:任務(wù)=1:2*/  if(p->queue_size>100){    int incr =0;    if(((float)p->rnum/p->queue_size) < THREAD_BUSY_PERCENT ){      incr = (p->queue_size*THREAD_BUSY_PERCENT) - p->rnum; /*計(jì)算需要增加線程個(gè)數(shù)*/      int i=0;      thread_info_t *tmp=NULL;      thread_pool_t *p = pool;      pthread_mutex_lock(&pool->queue_lock);      if(p->queue_size<100){        pthread_mutex_unlock(&pool->queue_lock);        return ;      }      for(i=0;i<incr;i++){        /*創(chuàng)建線程*/        tmp= (struct thread_info_s*)malloc(sizeof(struct thread_info_s));        if(tmp==NULL){          continue;        }else{          tmp->next = p->threads;          p->threads = tmp;        }        p->num ++;        p->rnum ++;        pthread_create(&(tmp->id),NULL,thread_excute_route,p);        tmp->state = THREAD_STATE_RUN;      }      pthread_mutex_unlock(&pool->queue_lock);    }  }  //pthread_cond_signal(&pool->extend_ready);}pthread_cond_t sum_ready;/*恢復(fù)初始線程個(gè)數(shù)*/void *thread_pool_is_need_recovery(void *arg){  thread_pool_t *pool = (thread_pool_t *)arg;  int i=0;  thread_info_t *tmp = NULL,*prev=NULL,*p1=NULL;  /*如果沒(méi)有任務(wù)了,當(dāng)前線程大于初始化的線程個(gè)數(shù)*/  while(1){    i=0;    if(pool->queue_size==0 && pool->rnum > pool->init_num ){      sleep(5);      /*5s秒內(nèi)還是這個(gè)狀態(tài)的話就,銷(xiāo)毀部分線程*/      if(pool->queue_size==0 && pool->rnum > pool->init_num ){        pthread_mutex_lock(&pool->queue_lock);        tmp = pool->threads;        while((pool->rnum != pool->init_num) && tmp){          /*找到空閑的線程*/          if(tmp->state != THREAD_STATE_TASK_PROCESSING){            i++;            if(prev)              prev->next  = tmp->next;            else              pool->threads = tmp->next;            pool->rnum --; /*正在運(yùn)行的線程減一*/            pool->knum ++; /*銷(xiāo)毀的線程加一*/            kill(tmp->id,SIGKILL); /*銷(xiāo)毀線程*/            p1 = tmp;            tmp = tmp->next;            free(p1);            continue;          }          prev = tmp;          tmp = tmp->next;        }        pthread_mutex_unlock(&pool->queue_lock);        printf("5s內(nèi)沒(méi)有新任務(wù)銷(xiāo)毀部分線程,銷(xiāo)毀了 %d 個(gè)線程/n",i);      }    }    sleep(5);  }}   /*打印一些信息的*/void *display_thread(void *arg){  thread_pool_t *p =(thread_pool_t *)arg;  thread_info_t *thread = NULL;  int i=0;  while(1){    printf("threads %d,running %d,killed %d/n",p->num,p->rnum,p->knum);  /*線程總數(shù),正在跑的,已銷(xiāo)毀的*/    thread = p->threads;    while(thread){      printf("id=%ld,state=%s/n",thread->id,thread_state_map[thread->state]);      thread = thread->next;    }    sleep(5);  }}