AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

一个Windows C++的线程池的实现

2013-09-18 1234
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： 此线程池所依赖的线程类，请参看《一个Windows C++的线程类实现》： http://blog.csdn.net/huyiyang2010/archive/2010/08/10/5801597.aspx   ThreadPoolExecutor.

此线程池所依赖的线程类，请参看《一个Windows C++的线程类实现》：

http://blog.csdn.net/huyiyang2010/archive/2010/08/10/5801597.aspx

 

ThreadPoolExecutor.h

[cpp]   view plain copy
  1. #ifndef __THREAD_POOL_EXECUTOR__  
  2. #define __THREAD_POOL_EXECUTOR__  
  3.   
  4. #include "Thread.h"  
  5. #include <set>  
  6. #include <list>  
  7. #include <windows.h>  
  8.   
  9. class CThreadPoolExecutor  
  10. {  
  11. public:  
  12.     CThreadPoolExecutor(void);  
  13.     ~CThreadPoolExecutor(void);  
  14.   
  15.     /** 
  16.       初始化线程池，创建minThreads个线程 
  17.     **/  
  18.     bool Init(unsigned int minThreads, unsigned int maxThreads, unsigned int maxPendingTaskse);  
  19.   
  20.     /** 
  21.       执行任务，若当前任务列表没有满，将此任务插入到任务列表，返回true 
  22.       若当前任务列表满了，但当前线程数量小于最大线程数，将创建新线程执行此任务，返回true 
  23.       若当前任务列表满了，但当前线程数量等于最大线程数，将丢弃此任务，返回false 
  24.     **/  
  25.     bool Execute(Runnable * pRunnable);  
  26.   
  27.     /** 
  28.       终止线程池，先制止塞入任务， 
  29.       然后等待直到任务列表为空， 
  30.       然后设置最小线程数量为0， 
  31.       等待直到线程数量为空， 
  32.       清空垃圾堆中的任务 
  33.     **/  
  34.     void Terminate();  
  35.   
  36.     /** 
  37.       返回线程池中当前的线程数量 
  38.     **/  
  39.     unsigned int GetThreadPoolSize();  
  40.   
  41. private:  
  42.     /** 
  43.       获取任务列表中的任务，若任务列表为空，返回NULL 
  44.     **/  
  45.     Runnable * GetTask();  
  46.   
  47.     static unsigned int WINAPI StaticThreadFunc(void * arg);  
  48.   
  49. private:  
  50.     class CWorker : public CThread  
  51.     {  
  52.     public:  
  53.         CWorker(CThreadPoolExecutor * pThreadPool, Runnable * pFirstTask = NULL);  
  54.         ~CWorker();  
  55.         void Run();  
  56.   
  57.     private:  
  58.         CThreadPoolExecutor * m_pThreadPool;  
  59.         Runnable * m_pFirstTask;  
  60.         volatile bool m_bRun;  
  61.     };  
  62.   
  63.     typedef std::set<CWorker *> ThreadPool;  
  64.     typedef std::list<Runnable *> Tasks;  
  65.     typedef Tasks::iterator TasksItr;  
  66.     typedef ThreadPool::iterator ThreadPoolItr;  
  67.   
  68.     ThreadPool m_ThreadPool;  
  69.     ThreadPool m_TrashThread;  
  70.     Tasks m_Tasks;  
  71.   
  72.     CRITICAL_SECTION m_csTasksLock;  
  73.     CRITICAL_SECTION m_csThreadPoolLock;  
  74.   
  75.     volatile bool m_bRun;  
  76.     volatile bool m_bEnableInsertTask;  
  77.     volatile unsigned int m_minThreads;  
  78.     volatile unsigned int m_maxThreads;  
  79.     volatile unsigned int m_maxPendingTasks;  
  80. };  
  81.   
  82. #endif  

 

 

ThreadPoolExecutor.cpp

[cpp]   view plain copy
  1. #include "ThreadPoolExecutor.h"  
  2.   
  3. CThreadPoolExecutor::CWorker::CWorker(CThreadPoolExecutor * pThreadPool, Runnable * pFirstTask) :   
  4. m_pThreadPool(pThreadPool),  
  5. m_pFirstTask(pFirstTask),  
  6. m_bRun(true)  
  7. {  
  8.       
  9. }  
  10.   
  11. CThreadPoolExecutor::CWorker::~CWorker()  
  12. {  
  13. }  
  14.   
  15. /** 
  16.   执行任务的工作线程。 
  17.   当前没有任务时， 
  18.   如果当前线程数量大于最小线程数量，减少线程， 
  19.   否则，执行清理程序，将线程类给释放掉 
  20. **/  
  21. void CThreadPoolExecutor::CWorker::Run()  
  22. {  
  23.     Runnable * pTask = NULL;  
  24.     while(m_bRun)  
  25.     {  
  26.         if(NULL == m_pFirstTask)  
  27.         {  
  28.             pTask = m_pThreadPool->GetTask();  
  29.         }  
  30.         else  
  31.         {  
  32.             pTask = m_pFirstTask;  
  33.             m_pFirstTask = NULL;  
  34.         }  
  35.   
  36.         if(NULL == pTask)  
  37.         {  
  38.             EnterCriticalSection(&(m_pThreadPool->m_csThreadPoolLock));  
  39.             if(m_pThreadPool->GetThreadPoolSize() > m_pThreadPool->m_minThreads)  
  40.             {  
  41.                 ThreadPoolItr itr = m_pThreadPool->m_ThreadPool.find(this);  
  42.                 if(itr != m_pThreadPool->m_ThreadPool.end())  
  43.                 {  
  44.                     m_pThreadPool->m_ThreadPool.erase(itr);  
  45.                     m_pThreadPool->m_TrashThread.insert(this);  
  46.                 }  
  47.                 m_bRun = false;  
  48.             }  
  49.             else  
  50.             {  
  51.                 ThreadPoolItr itr = m_pThreadPool->m_TrashThread.begin();  
  52.                 while(itr != m_pThreadPool->m_TrashThread.end())  
  53.                 {  
  54.                     (*itr)->Join();  
  55.                     delete (*itr);  
  56.                     m_pThreadPool->m_TrashThread.erase(itr);  
  57.                     itr = m_pThreadPool->m_TrashThread.begin();  
  58.                 }  
  59.             }  
  60.             LeaveCriticalSection(&(m_pThreadPool->m_csThreadPoolLock));  
  61.             continue;  
  62.         }  
  63.         else  
  64.         {  
  65.             pTask->Run();  
  66.             pTask = NULL;  
  67.         }  
  68.     }  
  69. }  
  70.   
  71. /////////////////////////////////////////////////////////////////////////////////////////////  
  72.   
  73. CThreadPoolExecutor::CThreadPoolExecutor(void) :   
  74. m_bRun(false),  
  75. m_bEnableInsertTask(false)  
  76. {  
  77.     InitializeCriticalSection(&m_csTasksLock);  
  78.     InitializeCriticalSection(&m_csThreadPoolLock);  
  79. }  
  80.   
  81. CThreadPoolExecutor::~CThreadPoolExecutor(void)  
  82. {  
  83.     Terminate();  
  84.     DeleteCriticalSection(&m_csTasksLock);  
  85.     DeleteCriticalSection(&m_csThreadPoolLock);  
  86. }  
  87.   
  88. bool CThreadPoolExecutor::Init(unsigned int minThreads, unsigned int maxThreads, unsigned int maxPendingTasks)  
  89. {  
  90.     if(minThreads == 0)  
  91.     {  
  92.         return false;  
  93.     }  
  94.     if(maxThreads < minThreads)  
  95.     {  
  96.         return false;  
  97.     }  
  98.     m_minThreads = minThreads;  
  99.     m_maxThreads = maxThreads;  
  100.     m_maxPendingTasks = maxPendingTasks;  
  101.     unsigned int i = m_ThreadPool.size();  
  102.     for(; i<minThreads; i++)  
  103.     {  
  104.         //创建线程  
  105.         CWorker * pWorker = new CWorker(this);  
  106.         if(NULL == pWorker)  
  107.         {  
  108.             return false;  
  109.         }  
  110.         EnterCriticalSection(&m_csThreadPoolLock);  
  111.         m_ThreadPool.insert(pWorker);  
  112.         LeaveCriticalSection(&m_csThreadPoolLock);  
  113.         pWorker->Start();  
  114.     }  
  115.     m_bRun = true;  
  116.     m_bEnableInsertTask = true;  
  117.     return true;  
  118. }  
  119.   
  120. bool CThreadPoolExecutor::Execute(Runnable * pRunnable)  
  121. {  
  122.     if(!m_bEnableInsertTask)  
  123.     {  
  124.         return false;  
  125.     }  
  126.     if(NULL == pRunnable)  
  127.     {  
  128.         return false;  
  129.     }  
  130.     if(m_Tasks.size() >= m_maxPendingTasks)  
  131.     {  
  132.         if(m_ThreadPool.size() < m_maxThreads)  
  133.         {  
  134.             CWorker * pWorker = new CWorker(this, pRunnable);  
  135.             if(NULL == pWorker)  
  136.             {  
  137.                 return false;  
  138.             }  
  139.             EnterCriticalSection(&m_csThreadPoolLock);  
  140.             m_ThreadPool.insert(pWorker);  
  141.             LeaveCriticalSection(&m_csThreadPoolLock);  
  142.             pWorker->Start();  
  143.         }  
  144.         else  
  145.         {  
  146.             return false;  
  147.         }  
  148.     }  
  149.     else  
  150.     {  
  151.         EnterCriticalSection(&m_csTasksLock);  
  152.         m_Tasks.push_back(pRunnable);  
  153.         LeaveCriticalSection(&m_csTasksLock);  
  154.     }  
  155.     return true;  
  156. }  
  157.   
  158. Runnable * CThreadPoolExecutor::GetTask()  
  159. {  
  160.     Runnable * Task = NULL;  
  161.     EnterCriticalSection(&m_csTasksLock);  
  162.     if(!m_Tasks.empty())  
  163.     {  
  164.         Task = m_Tasks.front();  
  165.         m_Tasks.pop_front();  
  166.     }  
  167.     LeaveCriticalSection(&m_csTasksLock);  
  168.     return Task;  
  169. }  
  170.   
  171. unsigned int CThreadPoolExecutor::GetThreadPoolSize()  
  172. {  
  173.     return m_ThreadPool.size();  
  174. }  
  175.   
  176. void CThreadPoolExecutor::Terminate()  
  177. {  
  178.     m_bEnableInsertTask = false;  
  179.     while(m_Tasks.size() > 0)  
  180.     {  
  181.         Sleep(1);  
  182.     }  
  183.     m_bRun = false;  
  184.     m_minThreads = 0;  
  185.     m_maxThreads = 0;  
  186.     m_maxPendingTasks = 0;  
  187.     while(m_ThreadPool.size() > 0)  
  188.     {  
  189.         Sleep(1);  
  190.     }  
  191.     EnterCriticalSection(&m_csThreadPoolLock);  
  192.     ThreadPoolItr itr = m_TrashThread.begin();  
  193.     while(itr != m_TrashThread.end())  
  194.     {  
  195.         (*itr)->Join();  
  196.         delete (*itr);  
  197.         m_TrashThread.erase(itr);  
  198.         itr = m_TrashThread.begin();  
  199.     }  
  200.     LeaveCriticalSection(&m_csThreadPoolLock);  
  201. }  

 

 

用法：

#include "Thread.h"
#include "ThreadPoolExecutor.h"

class R : public Runnable
{
public:
    ~R()
    {
    }
    void Run()
    {
        printf("Hello World/n");
    }
};

int _tmain(int argc, _TCHAR* argv[])
{
    CThreadPoolExecutor * pExecutor = new CThreadPoolExecutor();
    pExecutor->Init(1, 10, 50);
    R r;
    for(int i=0;i<100;i++)
    {
        while(!pExecutor->Execute(&r))
        {
        }
    }
    pExecutor->Terminate();
    delete pExecutor;
    getchar();
    return 0;
}

 

测试结果：

机器：

Intel(R) Core(TM)2 Duo CPU

E8400 @ 3.00GHz

2G内存

 

对于100个任务并且每个任务包含10000000个循环，任务中无等待：

单线程执行耗时：2281时间片

单线程池执行耗时：2219时间片

2个线程的线程池耗时：1156时间片

5个线程的线程池耗时：1166时间片

10个线程的线程池耗时：1157时间片

100个线程的线程池耗时：1177时间片

 

 

from:

文章标签：
C++
Windows
Java
关键词：
Windows C++
C++线程
Windows线程
C++线程池
C++ windows
double2li
目录
相关文章
游客akle7anmklvj6
|
5月前
|
存储 算法 C++
Windows共享文件：探秘C++实现的B树索引算法奇境
在数字化时代，Windows共享文件的高效管理至关重要。B树算法以其自平衡多路搜索特性，在文件索引与存储优化中表现出色。本文探讨B树在Windows共享文件中的应用，通过C++实现具体代码，展示其构建文件索引、优化数据存储的能力，提升文件检索效率。B树通过减少磁盘I/O操作，确保查询高效，为企业和个人提供流畅的文件共享体验。
游客akle7anmklvj6
179 3
石小浪♪
|
11月前
|
存储 消息中间件 资源调度
C++ 多线程之初识多线程
这篇文章介绍了C++多线程的基本概念，包括进程和线程的定义、并发的实现方式，以及如何在C++中创建和管理线程，包括使用`std::thread`库、线程的join和detach方法，并通过示例代码展示了如何创建和使用多线程。
石小浪♪
134 1
C++ 多线程之初识多线程
最好zzz
|
10月前
|
缓存 安全 C++
C++无锁队列：解锁多线程编程新境界
【10月更文挑战第27天】
最好zzz
618 7
潇洒洒
|
10月前
|
消息中间件 存储 安全
深入探索C++无锁队列：多线程编程的高效利器
【10月更文挑战第19天】
潇洒洒
872 6
小王老师呀
|
11月前
|
存储 并行计算 安全
C++多线程应用
【10月更文挑战第29天】C++ 中的多线程应用广泛，常见场景包括并行计算、网络编程中的并发服务器和图形用户界面（GUI）应用。通过多线程可以显著提升计算速度和响应能力。示例代码展示了如何使用 `pthread` 库创建和管理线程。注意事项包括数据同步与互斥、线程间通信和线程安全的类设计，以确保程序的正确性和稳定性。
小王老师呀
213 5
石小浪♪
|
11月前
|
存储 前端开发 C++
C++ 多线程之带返回值的线程处理函数
这篇文章介绍了在C++中使用`async`函数、`packaged_task`和`promise`三种方法来创建带返回值的线程处理函数。
石小浪♪
427 6
墨城coding
|
11月前
|
缓存 负载均衡 Java
c++写高性能的任务流线程池(万字详解！)
本文介绍了一种高性能的任务流线程池设计，涵盖多种优化机制。首先介绍了Work Steal机制，通过任务偷窃提高资源利用率。接着讨论了优先级任务，使不同优先级的任务得到合理调度。然后提出了缓存机制，通过环形缓存队列提升程序负载能力。Local Thread机制则通过预先创建线程减少创建和销毁线程的开销。Lock Free机制进一步减少了锁的竞争。容量动态调整机制根据任务负载动态调整线程数量。批量处理机制提高了任务处理效率。此外，还介绍了负载均衡、避免等待、预测优化、减少复制等策略。最后，任务组的设计便于管理和复用多任务。整体设计旨在提升线程池的性能和稳定性。
墨城coding
252 5
石小浪♪
|
11月前
|
C++
C++ 多线程之线程管理函数
这篇文章介绍了C++中多线程编程的几个关键函数，包括获取线程ID的`get_id()`，延时函数`sleep_for()`，线程让步函数`yield()`，以及阻塞线程直到指定时间的`sleep_until()`。
石小浪♪
151 0
C++ 多线程之线程管理函数
石小浪♪
|
11月前
|
资源调度 Linux 调度
Linux C/C++之线程基础
这篇文章详细介绍了Linux下C/C++线程的基本概念、创建和管理线程的方法，以及线程同步的各种机制，并通过实例代码展示了线程同步技术的应用。
石小浪♪
170 0
Linux C/C++之线程基础
路边两盏灯
|
Java Windows
【Azure Developer】Windows中通过pslist命令查看到Java进程和线程信息，但为什么和代码中打印出来的进程号不一致呢？
【Azure Developer】Windows中通过pslist命令查看到Java进程和线程信息，但为什么和代码中打印出来的进程号不一致呢？
路边两盏灯
157 1

热门文章

最新文章

  • 1
    C++语言中指针变量int和取值操作ptr详细说明。
  • 2
    C++面试周刊(3):面试不慌,这样回答指针与引用，青铜秒变王者
  • 3
    windows 11 系统安装教程，包括U盘和本机安装，附windows 11各版本ISO镜像
  • 4
    VirtualBox 7.2.0 (macOS, Linux, Windows) - 开源跨平台虚拟化软件
  • 5
    双系统必备：Ubuntu自动挂载Windows分区
  • 6
    Tenable Nessus 10.9.3 (macOS, Linux, Windows) - 漏洞评估解决方案
  • 7
    Nexpose 8.18.0 for Linux & Windows - 漏洞扫描
  • 8
    C++开源游戏推荐，雷神之锤1/2/3
  • 9
    2015级C++第12周实践项目 运算符重载（一）
  • 10
    13、C++ Primer 4th 笔记，泛型算法
  • 1
    Windows就是Linux：WSL带来全新Linux开发体验
    935
  • 2
    electron程序运行在某些 windows 上白屏
    352
  • 3
    windows配置supervisor实现nginx自启
    309
  • 4
    [windows]远程桌面失败提示CredSSP加密修正
    2256
  • 5
    Windows查找监听端口对应的进程及其路径
    287
  • 6
    Windows安装hexo并配置nginx
    149
  • 7
    WPF在.NET9中的重大更新：Windows 11 主题
    198
  • 8
    【开源推荐】windows窗口管理器
    296
  • 9
    解锁QtCreator跨界神技！Windows下轻松驾驭OpenCV动态库，让你的跨平台开发如虎添翼，秒变视觉编程大师！
    616
  • 10
    从Windows到Linux，Python系统调用如何让代码飞翔🚀
    133

    • 相关课程

    更多
  • C++ 入门教程开发文档

    • 相关电子书

    更多
  • 使用C++11开发PHP7扩展
  • GPON Class C++ SFP O;T Transce
  • GPON Class C++ SFP OLT Transce
    • 下一篇
    蛋白质语言模型 ProGen：在实验室合成由 AI 预测的蛋白质