博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
Linux C++ 实现线程池
阅读量:4095 次
发布时间:2019-05-25

本文共 4975 字,大约阅读时间需要 16 分钟。

http://blog.csdn.net/qq_25425023/article/details/53914609

线程池中的线程,在任务队列为空的时候,等待任务的到来,任务队列中有任务时,则依次获取任务来执行,任务队列需要同步。

  Linux线程同步有多种方法:互斥量、信号量、条件变量等。

  下面是根据互斥量、信号量、条件变量封装的三个类。

  线程池中用到了互斥量和信号量。

 

[cpp] 
 
  1. #ifndef _LOCKER_H_  
  2. #define _LOCKER_H_  
  3.   
  4. #include <pthread.h>  
  5. #include <stdio.h>  
  6. #include <semaphore.h>  
  7.   
  8. /*信号量的类*/  
  9. class sem_locker  
  10. {  
  11. private:  
  12.     sem_t m_sem;  
  13.   
  14. public:  
  15.     //初始化信号量  
  16.     sem_locker()  
  17.     {  
  18.     if(sem_init(&m_sem, 0, 0) != 0)  
  19.         printf("sem init error\n");  
  20.     }  
  21.     //销毁信号量  
  22.     ~sem_locker()  
  23.     {  
  24.     sem_destroy(&m_sem);  
  25.     }  
  26.   
  27.     //等待信号量  
  28.     bool wait()  
  29.     {  
  30.     return sem_wait(&m_sem) == 0;  
  31.     }  
  32.     //添加信号量  
  33.     bool add()  
  34.     {  
  35.     return sem_post(&m_sem) == 0;  
  36.     }  
  37. };  
  38.   
  39.   
  40. /*互斥 locker*/  
  41. class mutex_locker  
  42. {  
  43. private:  
  44.     pthread_mutex_t m_mutex;  
  45.   
  46. public:  
  47.     mutex_locker()  
  48.     {  
  49.         if(pthread_mutex_init(&m_mutex, NULL) != 0)  
  50.         printf("mutex init error!");  
  51.     }  
  52.     ~mutex_locker()  
  53.     {  
  54.     pthread_mutex_destroy(&m_mutex);  
  55.     }  
  56.   
  57.     bool mutex_lock()  //lock mutex  
  58.     {  
  59.     return pthread_mutex_lock(&m_mutex) == 0;  
  60.     }  
  61.     bool mutex_unlock()   //unlock  
  62.     {  
  63.     return pthread_mutex_unlock(&m_mutex) == 0;  
  64.     }  
  65. };  
  66.   
  67. /*条件变量 locker*/  
  68. class cond_locker  
  69. {  
  70. private:  
  71.     pthread_mutex_t m_mutex;  
  72.     pthread_cond_t m_cond;  
  73.   
  74. public:  
  75.     // 初始化 m_mutex and m_cond  
  76.     cond_locker()  
  77.     {  
  78.     if(pthread_mutex_init(&m_mutex, NULL) != 0)  
  79.         printf("mutex init error");  
  80.     if(pthread_cond_init(&m_cond, NULL) != 0)  
  81.     {   //条件变量初始化是被,释放初始化成功的mutex  
  82.         pthread_mutex_destroy(&m_mutex);  
  83.         printf("cond init error");  
  84.     }  
  85.     }  
  86.     // destroy mutex and cond  
  87.     ~cond_locker()  
  88.     {  
  89.     pthread_mutex_destroy(&m_mutex);  
  90.     pthread_cond_destroy(&m_cond);  
  91.     }  
  92.     //等待条件变量  
  93.     bool wait()  
  94.     {  
  95.     int ans = 0;  
  96.     pthread_mutex_lock(&m_mutex);  
  97.     ans = pthread_cond_wait(&m_cond, &m_mutex);  
  98.     pthread_mutex_unlock(&m_mutex);  
  99.     return ans == 0;  
  100.     }  
  101.     //唤醒等待条件变量的线程  
  102.     bool signal()  
  103.     {  
  104.     return pthread_cond_signal(&m_cond) == 0;  
  105.     }  
  106.   
  107. };  
  108.   
  109. #endif  

下面的是线程池类,是一个模版类:

[cpp] 
 
  1. #ifndef _PTHREAD_POOL_  
  2. #define _PTHREAD_POOL_  
  3.   
  4. #include "locker.h"  
  5. #include <list>  
  6. #include <stdio.h>  
  7. #include <exception>  
  8. #include <errno.h>  
  9. #include <pthread.h>  
  10. #include <iostream>  
  11.   
  12. template<class T>  
  13. class threadpool  
  14. {  
  15. private:  
  16.     int thread_number;  //线程池的线程数  
  17.     int max_task_number;  //任务队列中的最大任务数  
  18.     pthread_t *all_threads;   //线程数组  
  19.     std::list<T *> task_queue; //任务队列  
  20.     mutex_locker queue_mutex_locker;  //互斥锁  
  21.     sem_locker queue_sem_locker;   //信号量  
  22.     bool is_stop; //是否结束线程  
  23. public:  
  24.     threadpool(int thread_num = 20, int max_task_num = 30);  
  25.     ~threadpool();  
  26.     bool append_task(T *task);  
  27.     void start();  
  28.     void stop();  
  29. private:  
  30.     //线程运行的函数。执行run()函数  
  31.     static void *worker(void *arg);  
  32.     void run();  
  33. };  
  34.   
  35. template <class T>  
  36. threadpool<T>::threadpool(int thread_num, int max_task_num):  
  37.     thread_number(thread_num), max_task_number(max_task_num),  
  38.     is_stop(false), all_threads(NULL)  
  39. {  
  40.     if((thread_num <= 0) || max_task_num <= 0)  
  41.     printf("threadpool can't init because thread_number = 0"  
  42.         " or max_task_number = 0");  
  43.   
  44.     all_threads = new pthread_t[thread_number];  
  45.     if(!all_threads)  
  46.         printf("can't init threadpool because thread array can't new");  
  47. }  
  48.   
  49. template <class T>  
  50. threadpool<T>::~threadpool()  
  51. {  
  52.     delete []all_threads;  
  53.     is_stop = true;  
  54. }  
  55.   
  56. template <class T>  
  57. void threadpool<T>::stop()  
  58. {  
  59.     is_stop = true;  
  60.     //queue_sem_locker.add();  
  61. }  
  62.   
  63. template <class T>  
  64. void threadpool<T>::start()  
  65. {  
  66.     for(int i = 0; i < thread_number; ++i)  
  67.     {  
  68.     printf("create the %dth pthread\n", i);  
  69.     if(pthread_create(all_threads + i, NULL, worker, this) != 0)  
  70.     {
    //创建线程失败,清除成功申请的资源并抛出异常  
  71.         delete []all_threads;  
  72.         throw std::exception();  
  73.     }  
  74.     if(pthread_detach(all_threads[i]))  
  75.     {
    //将线程设置为脱离线程,失败则清除成功申请的资源并抛出异常  
  76.         delete []all_threads;  
  77.         throw std::exception();  
  78.     }  
  79.     }  
  80. }  
  81. //添加任务进入任务队列  
  82. template <class T>  
  83. bool threadpool<T>::append_task(T *task)  
  84. {   //获取互斥锁  
  85.     queue_mutex_locker.mutex_lock();  
  86.     //判断队列中任务数是否大于最大任务数  
  87.     if(task_queue.size() > max_task_number)  
  88.     {
    //是则释放互斥锁  
  89.     queue_mutex_locker.mutex_unlock();  
  90.     return false;  
  91.     }  
  92.     //添加进入队列  
  93.     task_queue.push_back(task);  
  94.     queue_mutex_locker.mutex_unlock();  
  95.     //唤醒等待任务的线程  
  96.     queue_sem_locker.add();  
  97.     return true;  
  98. }  
  99.   
  100. template <class T>  
  101. void *threadpool<T>::worker(void *arg)  
  102. {  
  103.     threadpool *pool = (threadpool *)arg;  
  104.     pool->run();  
  105.     return pool;  
  106. }  
  107.   
  108. template <class T>  
  109. void threadpool<T>::run()  
  110. {  
  111.     while(!is_stop)  
  112.     {   //等待任务  
  113.     queue_sem_locker.wait();      
  114.     if(errno == EINTR)  
  115.     {  
  116.         printf("errno");  
  117.         continue;  
  118.     }  
  119.     //获取互斥锁  
  120.     queue_mutex_locker.mutex_lock();  
  121.     //判断任务队列是否为空  
  122.     if(task_queue.empty())  
  123.     {  
  124.         queue_mutex_locker.mutex_unlock();  
  125.         continue;  
  126.     }  
  127.     //获取队头任务并执行  
  128.     T *task = task_queue.front();  
  129.     task_queue.pop_front();  
  130.     queue_mutex_locker.mutex_unlock();  
  131.     if(!task)  
  132.         continue;  
  133. //  printf("pthreadId = %ld\n", (unsigned long)pthread_self());   
  134.     task->doit();  //doit是T对象中的方法  
  135.     }  
  136.     //测试用  
  137.     printf("close %ld\n", (unsigned long)pthread_self());  
  138. }  
  139.   
  140. #endif  

以上参考《Linux高性能服务器编程》

写个程序对线程池进行测试:

[cpp] 
 
  1. #include <stdio.h>  
  2. #include <iostream>  
  3. #include <unistd.h>  
  4.   
  5. #include "thread_pool.h"  
  6.   
  7. class task  
  8. {  
  9. private:  
  10.     int number;  
  11.   
  12. public:  
  13.     task(int num) : number(num)  
  14.     {  
  15.     }  
  16.     ~task()  
  17.     {  
  18.     }  
  19.   
  20.     void doit()  
  21.     {  
  22.     printf("this is the %dth task\n", number);  
  23.     }  
  24. };  
  25.   
  26. int main()  
  27. {  
  28.     task *ta;  
  29.     threadpool<task> pool(10, 15);  
  30. //    pool.start();  
  31.     for(int i = 0; i < 20; ++i)  
  32.     {  
  33.     ta = new task(i);  
  34. //  sleep(2);  
  35.     pool.append_task(ta);  
  36.     }  
  37.     pool.start();  
  38.     sleep(10);  
  39.     printf("close the thread pool\n");  
  40.     pool.stop();  
  41.     pause();  
  42.     return 0;  
  43. }  

经测试,线程池可以正常使用。

下一篇博客,使用线程池来实现回射服务器,测试可以达到多大的并发量。

你可能感兴趣的文章
2017年,这一次我们不聊技术
查看>>
实现接口创建线程
查看>>
Java对象序列化与反序列化(1)
查看>>
HTML5的表单验证实例
查看>>
JavaScript入门笔记:全选功能的实现
查看>>
程序设计方法概述:从面相对象到面向功能到面向对象
查看>>
数据库事务
查看>>
JavaScript基础1:JavaScript 错误 - Throw、Try 和 Catch
查看>>
SQL基础总结——20150730
查看>>
SQL join
查看>>
JavaScript实现页面无刷新让时间走动
查看>>
CSS实例:Tab选项卡效果
查看>>
前端设计之特效表单
查看>>
前端设计之CSS布局:上中下三栏自适应高度CSS布局
查看>>
Java的时间操作玩法实例若干
查看>>
JavaScript:时间日期格式验证大全
查看>>
pinyin4j:拼音与汉字的转换实例
查看>>
XML工具代码:SAX从String字符串XML内获取指定节点或属性的值
查看>>
时间日期:获取两个日期相差几天
查看>>
责任链模式 Chain of Responsibility
查看>>