睿豐德科技 專注RFID識別技術和條碼識別技術與管理軟件的集成項目。質量追溯系統、MES系統、金蝶與條碼系統對接、用友與條碼系統對接

此線程池所依賴的線程類，請參看《一個Windows C++的線程類實現》：

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

 

ThreadPoolExecutor.h

[cpp] view plaincopy

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 plaincopy

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:

RFID管理系統集成商 RFID中間件 條碼系統中間層 物聯網軟件集成