线程池设计案例一

930阅读　0评论2016-02-18　taohorse
分类：C/C++

一、适用场景

    首先，必须明确一点，线程池不是万能的，它有其特定的使用场景。使用线程池是为了减小线程本身的开销对应用性能所产生的影响，但是其前提是线程本身创建、销毁的开销和线程执行任务的开销相比是不可忽略的。如果线程本身创建、销毁的开销对应用程序的性能可以忽略不计，那么使用/不使用线程池对程序的性能并不会有太大的影响。

    线程池通常适合以下几种场景：

        ①、单位时间内处理的任务频繁，且任务时间较短

        ②、对实时性要求较高。如果接收到任务之后再创建线程，可能无法满足实时性的要求，此时必须使用线程池。

        ③、必须经常面对高突发性事件。比如Web服务器。如果有足球转播，则服务器将产生巨大冲击，此时使用传统方法，则必须不停的大量创建、销毁线程。此时采用动态线程池可以避免这种情况的发生。

二、代码实现

注意事项：

之 前使用的是非分离线程，当存在线程异常退出时，操作系统无法及时回收内存空间。为了解决此问题，现使用分离线程。但是必须注意：不要使用 pthread_detach()来使线程成为分离线程，而应该通过线程属性（pthread_attr_t）的参数来设置线程为分离线程。否则，当线程 退出后，再调用pthread_kill（）来判断线程是否还存在时，很可能出现段错误。


2.1 头文件

  1. #if !defined(__THREAD_POOL_H__)  
  2. #define __THREAD_POOL_H__  
  3.   
  4. #include   
  5. #include   
  6. #include   
  7. #include   
  8. #include   
  9. #include   
  10. #include   
  11.   
  12. // 布尔类型  
  13. typedef int bool;  
  14. #define false (0)  
  15. #define true  (1)  
  16.   
  17. /* 线程任务链表 */  
  18. typedef struct _thread_worker_t  
  19. {  
  20.     void *(*process)(void *arg);  /* 线程处理的任务 */  
  21.     void *arg;                    /* 任务接口参数 */  
  22.     struct _thread_worker_t *next;/* 下一个节点 */  
  23. }thread_worker_t;  
  24.   
  25. /* 线程池对象 */  
  26. typedef struct  
  27. {  
  28.     pthread_mutex_t queue_lock;   /* 队列互斥锁 */  
  29.     pthread_cond_t queue_ready;   /* 队列条件锁 */  
  30.   
  31.     thread_worker_t *head;        /* 任务队列头指针 */  
  32.     bool isdestroy;               /* 是否已销毁线程 */  
  33.     pthread_t *threadid;          /* 线程ID数组 —动态分配空间 */  
  34.     int reqnum;                   /* 请求创建的线程个数 */  
  35.     int num;                      /* 实际创建的线程个数 */  
  36.     int queue_size;               /* 工作队列当前大小 */  
  37. }thread_pool_t;  
  38.   
  39. /* 函数声明 */  
  40. extern int thread_pool_init(thread_pool_t **pool, int num);  
  41. extern int thread_pool_add_worker(thread_pool_t *pool, void *(*process)(void *arg), void *arg);  
  42. extern int thread_pool_keepalive(thread_pool_t *pool);  
  43. extern int thread_pool_destroy(thread_pool_t *pool);  
  44.   
  45. #endif /*__THREAD_POOL_H__*/ 


2.2 函数实现
  1. /************************************************************* 
  2.  **功  能：线程池的初始化 
  3.  **参  数： 
  4.  **    pool：线程池对象 
  5.  **    num ：线程池中线程个数 
  6.  **返回值：0：成功 !0: 失败 
  7.  *************************************************************/  
  8. int thread_pool_init(thread_pool_t **pool, int num)  
  9. {  
  10.     int idx = 0;  
  11.   
  12.     /* 为线程池分配空间 */  
  13.     *pool = (thread_pool_t*)calloc(1, sizeof(thread_pool_t));  
  14.     if(NULL == *pool)  
  15.     {  
  16.         return -1;  
  17.     }  
  18.   
  19.     /* 初始化线程池 */  
  20.     pthread_mutex_init(&((*pool)->queue_lock), NULL);  
  21.     pthread_cond_init(&((*pool)->queue_ready), NULL);  
  22.     (*pool)->head = NULL;  
  23.     (*pool)->reqnum = num;  
  24.     (*pool)->queue_size = 0;  
  25.     (*pool)->isdestroy = false;  
  26.     (*pool)->threadid = (pthread_t*)calloc(1, num*sizeof(pthread_t));  
  27.     if(NULL == (*pool)->threadid)  
  28.     {  
  29.         free(*pool);  
  30.         (*pool) = NULL;  
  31.   
  32.         return -1;  
  33.     }  
  34.   
  35.     /* 依次创建线程 */  
  36.     for(idx=0; idx
  37.     {  
  38.         ret = thread_create_detach(*pool, idx);  
  39.         if(0 != ret)  
  40.         {  
  41.             return -1;  
  42.         }  
  43.         (*pool)->num++;  
  44.     }  
  45.   
  46.     return 0;  
  47. }  

  1. /************************************************************* 
  2.  **功  能：将任务加入线程池处理队列 
  3.  **参  数： 
  4.  **    pool：线程池对象 
  5.  **    process：需处理的任务 
  6.  **    arg: process函数的参数 
  7.  **返回值：0：成功 !0: 失败 
  8.  *************************************************************/  
  9. int thread_pool_add_worker(thread_pool_t *pool, void *(*process)(void *arg), void *arg)  
  10. {  
  11.     thread_worker_t *worker=NULL, *member=NULL;  
  12.       
  13.     worker = (thread_worker_t*)calloc(1, sizeof(thread_worker_t));  
  14.     if(NULL == worker)  
  15.     {  
  16.         return -1;  
  17.     }  
  18.   
  19.     worker->process = process;  
  20.     worker->arg = arg;  
  21.     worker->next = NULL;  
  22.   
  23.     pthread_mutex_lock(&(pool->queue_lock));  
  24.   
  25.     member = pool->head;  
  26.     if(NULL != member)  
  27.     {  
  28.         while(NULL != member->next) member = member->next;  
  29.         member->next = worker;  
  30.     }  
  31.     else  
  32.     {  
  33.         pool->head = worker;  
  34.     }  
  35.   
  36.     pool->queue_size++;  
  37.   
  38.     pthread_mutex_unlock(&(pool->queue_lock));  
  39.     pthread_cond_signal(&(pool->queue_ready));  
  40.   
  41.     return 0;  
  42. }  

  1. /****************************************************************************** 
  2.  **函数名称: thread_pool_keepalive 
  3.  **功    能: 线程保活 
  4.  **输入参数:  
  5.  **       pool: 线程池 
  6.  **输出参数: NONE 
  7.  **返    回: 0: success !0: failed 
  8.  **实现过程: 
  9.  **      1. 判断线程是否存在 
  10.  **      2. 不存在，说明线程死亡，需重新创建 
  11.  ******************************************************************************/  
  12. int thread_pool_keepalive(thread_pool_t *pool)  
  13. {  
  14.     int idx=0, ret=0;  
  15.   
  16.     for(idx=0; idxnum; idx++)  
  17.     {  
  18.         ret = pthread_kill(pool->thread[idx], 0);  
  19.         if(ESRCH == ret)  
  20.         {  
  21.             ret = thread_create_detach(pool, idx);  
  22.             if(ret < 0)  
  23.             {  
  24.                 return -1;  
  25.             }  
  26.         }  
  27.     }  
  28.   
  29.     return 0;  
  30. } 

  1. /************************************************************* 
  2.  **功  能：线程池的销毁 
  3.  **参  数： 
  4.  **    pool：线程池对象 
  5.  **返回值：0：成功 !0: 失败 
  6.  *************************************************************/  
  7. int thread_pool_destroy(thread_pool_t *pool)  
  8. {  
  9.     int idx = 0;  
  10.     thread_worker_t *member = NULL;  
  11.   
  12.     if(false != pool->isdestroy)  
  13.     {  
  14.         return -1;  
  15.     }  
  16.   
  17.     pool->isdestroy = true;  
  18.   
  19.     pthread_cond_broadcast(&(pool->queue_ready));  
  20.     for(idx=0; idxnum; idx++)  
  21.     {  
  22.         ret = pthread_kill(pool->threadid[idx], 0);  
  23.         if(ESRCH == ret)  
  24.         {  
  25.             continue;  
  26.         }  
  27.         else  
  28.         {  
  29.             idx--;  
  30.             sleep(1);  
  31.         }  
  32.     }  
  33.   
  34.     free(pool->threadid);  
  35.     pool->threadid = NULL;  
  36.   
  37.     while(NULL != pool->head)  
  38.     {  
  39.         member = pool->head;  
  40.         pool->head = member->next;  
  41.         free(member);  
  42.     }  
  43.   
  44.     pthread_mutex_destroy(&(pool->queue_lock));  
  45.     pthread_cond_destroy(&(pool->queue_ready));  
  46.     free(pool);  
  47.       
  48.     return 0;  
  49. }  

  1. /****************************************************************************** 
  2.  **函数名称: thread_create_detach 
  3.  **功    能: 创建分离线程 
  4.  **输入参数:  
  5.  **       pool: 线程池 
  6.  **       idx: 线程索引号 
  7.  **输出参数: NONE 
  8.  **返    回: 0: success !0: failed 
  9.  ******************************************************************************/  
  10. static int thread_create_detach(thread_pool_t *pool, int idx)  
  11. {  
  12.     int ret = 0;  
  13.     pthread_attr_t attr;  
  14.   
  15.     do  
  16.     {  
  17.         ret = pthread_attr_init(&attr);  
  18.         if(0 != ret)  
  19.         {  
  20.             return -1;  
  21.         }  
  22.           
  23.         ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);  
  24.         if(0 != ret)  
  25.         {  
  26.             return -1;  
  27.         }  
  28.           
  29.         ret = pthread_create(&((*pool)->threadid[idx]), &attr, thread_routine, *pool);  
  30.         if(0 != ret)  
  31.         {  
  32.             pthread_attr_destroy(&attr);  
  33.               
  34.             if(EINTR == errno)  
  35.             {  
  36.                 continue;  
  37.             }  
  38.             return -1;  
  39.         }  
  40.         pthread_attr_destroy(&attr);  
  41.     }while(0);  
  42.   
  43.     return 0;  
  44. }  

  1. /************************************************************* 
  2.  **功  能：线程池各个线程入口函数 
  3.  **参  数： 
  4.  **    arg：线程池对象 
  5.  **返回值：0：成功 !0: 失败 
  6.  *************************************************************/  
  7. static void *thread_routine(void *arg)  
  8. {  
  9.     thread_worker_t *worker = NULL;  
  10.     thread_pool_t *pool = (thread_pool_t*)arg;  
  11.   
  12.     while(1)  
  13.     {  
  14.         pthread_mutex_lock(&(pool->queue_lock));  
  15.         while((false == pool->isdestroy) && (0 == pool->queue_size))  
  16.         {  
  17.             pthread_cond_wait(&(pool->queue_ready), &(pool->queue_lock));  
  18.         }  
  19.   
  20.         if(false != pool->isdestroy)  
  21.         {  
  22.             pthread_mutex_unlock(&(pool->queue_lock));  
  23.             pthread_exit(NULL);  
  24.         }  
  25.   
  26.         pool->queue_size--;  
  27.         worker = pool->head;  
  28.         pool->head = worker->next;  
  29.         pthread_mutex_unlock(&(pool->queue_lock);  
  30.         /* 执行队列中的任务 */  
  31.         (*(worker->process))(worker->arg);  
  32.   
  33.         free(worker);  
  34.         worker = NULL;  
  35.     }  
  36. }  


  1. #define THREAD_MAX_NUM (32)  
  2. #define SLEEP   (10)  
  3.   
  4. int myprocess(void *arg)  
  5. {  
  6.     fprintf(stdout, "[%s][%d] threadid:%d arg:%d", __FILE__, __LINE__, pthread_self(), *(int*)arg);  
  7.     return 0;  
  8. }  
  9.   
  10. int main(void)  
  11. {  
  12.     int ret=0, idx=0;  
  13.     thread_pool_t *pool = NULL;  
  14.     int array[THREAD_MAX_NUM] = {0};  
  15.   
  16.     ret = thread_pool_init(&pool, THREAD_MAX_NUM);  
  17.     if(ret < 0)  
  18.     {  
  19.         return -1;  
  20.     }  
  21.   
  22.     for(idx=0; idx
  23.     {  
  24.         array[idx] = idx;  
  25.         thread_pool_add_worker(pool, myprocess, &array[idx]); /* 注意：地址各不相同 */  
  26.     }  
  27.   
  28.     thread_pool_destroy(pool);  
  29.     pool = NULL;  
  30.     return 0;  
  31. } 



上一篇：安装Twisted报fatal error: Python.h: 没有那个文件或目录
下一篇：关于一些url中传递参数有空格问题