這里給大家介紹了C++中對于pthread線程的一個簡單應用以及使用繼承CDoit，實現其中的start和end，有需要的小伙伴可以參考下

這是對pthread線程的一個簡單應用

1. 實現了線程池的概念，線程可以重復使用。

2. 對信號量，互斥鎖等進行封裝，業務處理函數中只需寫和業務相關的代碼。

3. 移植性好。如果想把這個線程池代碼應用到自己的實現中去，只要寫自己的業務處理函數和改寫工作隊列數據的處理方法就可以了。

Sample代碼主要包括一個主程序和兩個線程實現類

ThreadTest.cpp：主程序

CThreadManager：線程管理Class，線程池的實現類

CThread：線程Class.

主程序實現方法。

1. 實現main函數和一個需要線程處理的業務函數(例子代碼中業務函數是一個簡單的計算函數Count)。在main函數中創建CThreadManager的實例，產生線程池。這個時候，把業務函數作為函數指針傳到CThreadManager里面，最終會被線程調用。

2. 向工作隊列中放入業務函數要處理的數據。

3. 設置信號量，喚醒線程。

1. // 線程要執行的函數 
2. int Count(int nWork) 
3. { 
4. int nResult = nWork * nWork; 
5. printf("count result is %d/n",nResult); 
6. return 0; 
7. } 
8. int main() { 
9. // 創建線程管理類的實例，把要執行的線程函數和最大線程數傳進去 
10. CThreadManager* pManager = new CThreadManager(Count, 3); 
11. // 把要進行計算的數放到工作隊列中 
12. pManager->PushWorkQue(5); 
13. pManager->PushWorkQue(20); 
14. // 設置信號量，喚醒線程 
15. pManager->PostSem(); 
16. pManager->PostSem(); 
17. // 等待子線程執行 
18. sleep(1); 
19. return 0; 
20. } 

CThreadManager實現的方法

1. 把信號量和互斥鎖等封裝成自己的函數

2. 在new方法里，循環調用CThread的new方法，啟動一定數量(可設定)的線程，產生線程池。

3. 這些線程啟動后，就會執行CThreadManager中的ManageFuction函數。這個函數是無限循環的，保證了線程在整個程序的生命周期中不銷毀。

4. 在循環處理里面，第一行代碼就是等待一個信號量，這個信號量是由主程序進行設置的，這個信號信號量如果沒有被設置(代表暫時沒有需要處理的工作)，所有線程都在這里阻塞著。

4. 一旦信號量被設置，根據Linux線程調度機制，在阻塞的線程隊列中，其中一個線程被喚醒，可以執行后面的代碼。

5. 從工作隊列中取出要進行處理的數據(使用互斥鎖進行排他)

6. 通過函數指針調用main函數傳過來的業務函數，處理數據。

7. 業務函數執行完之后，線程進入下一個循環，等待新的信號量。

1. class CThreadManager { 
2. friend void* ManageFuction(void*); 
3. private: 
4. sem_t m_sem; // 信號量 
5. pthread_mutex_t m_mutex; // 互斥鎖 
6. queue<int> m_queWork; // 工作隊列 
7. list<CThread*> m_lstThread; // 線程list 
8. int (*m_threadFuction)(int); //函數指針，指向main函數傳過來的線程執行函數 
9. public: 
10. CThreadManager(int (*threadFuction)(int), int nMaxThreadCnt); 
11. virtual ~CThreadManager(); 
12. int WaitSem(); 
13. int PostSem(); 
14. int LockMutex(); 
15. int UnlockMutex(); 
16. void PushWorkQue(int nWork); 
17. int PopWorkQue(); 
18. int RunThreadFunction(int nWork); 
19. }; 
20. // 線程執行函數，它只是個殼子，處理信號量和互斥鎖等， 
21. // 最后調用main函數傳過來的線程執行函數來實現業務處理 
22. void* ManageFuction(void* argv) 
23. { 
24. CThreadManager* pManager = (CThreadManager*)argv; 
25. // 進行無限循環（意味著線程是不銷毀的，重復利用） 
26. while(true) 
27. { 
28. // 線程開啟后，就在這里阻塞著，直到main函數設置了信號量 
29. pManager->WaitSem(); 
30. printf("thread wakeup./n"); 
31. // 從工作隊列中取出要處理的數 
32. pManager->LockMutex(); 
33. int nWork = pManager->PopWorkQue(); 
34. pManager->UnlockMutex(); 
35. printf("call Count function./n"); 
36. pManager->RunThreadFunction(nWork); 
37. } 
38. return 0; 
39. } 
40. // 構造方法 
41. CThreadManager::CThreadManager(int (*threadFuction)(int), int nMaxThreadCnt)  
42. { 
43. sem_init(&m_sem, 0, 0); 
44. pthread_mutex_init(&m_mutex, NULL); 
45. m_threadFuction = threadFuction; 
46. for(int i=0; i<nMaxThreadCnt; i++) 
47. { 
48. CThread* pThread = new CThread(ManageFuction, this); 
49. printf("thread started./n"); 
50. m_lstThread.push_back(pThread); 
51. } 
52. } 

CThread實現的方法

CThreadManager比較簡單，封裝了創建線程和join線程的函數。

1. CThread::CThread(void* (*threadFuction)(void*),void* threadArgv)  
2. { 
3. // 初始化線程屬性 
4. pthread_attr_t threadAttr; 
5. pthread_attr_init(&threadAttr); 
6.  
7. pthread_create(&m_thread, &threadAttr, threadFuction, threadArgv); 
8. } 

c++線程池，繼承CDoit，實現其中的start和end

1. /* 
2. * 多線程管理類 
3. *  
4. */ 
5.  
6. #ifndef CTHREADPOOLMANAGE_H 
7. #define CTHREADPOOLMANAGE_H 
8. #include <iostream> 
9. #include <pthread.h> 
10. #include <unistd.h>  
11. #include <list> 
12. #include <vector> 
13. #include <time.h> 
14. #include <asm/errno.h> 
15.  
16. #define USLEEP_TIME 100 
17. #define CHECK_TIME 1 
18.  
19.  
20. using namespace std; 
21. class CDoit 
22. { 
23. public: 
24. virtual int start(void *){}; 
25. virtual int end(){}; 
26. }; 
27.  
28.  
29. class CthreadPoolManage 
30. { 
31. private: 
32. int _minThreads; //最少保留幾個線程 
33. int _maxThreads; //最多可以有幾個線程 
34. int _waitSec; //空閑多少秒后將線程關閉 
35. class threadInfo{ 
36. public: 
37. threadInfo(){ 
38. isbusy = false;  
39. doFlag = true; 
40. } 
41. // 
42. pthread_mutex_t mtx=PTHREAD_MUTEX_INITIALIZER; 
43. pthread_cond_t cond=PTHREAD_COND_INITIALIZER; 
44. bool isbusy; //是否空閑 
45. bool doFlag; 
46. // 
47. time_t beginTime; //線程不工作開始時間 
48. pthread_t cThreadPid; //線程id 
49. pthread_attr_t cThreadAttr; //線程屬性 
50. CDoit * doit; //任務類 
51. void * value; //需要傳遞的值 
52. }; 
53. //線程函數 
54. static void* startThread(void*); 
55. //任務隊列鎖 
56. pthread_mutex_t _duty_mutex; 
57. //任務隊列 
58. list<threadInfo*> _dutyList; 
59. //線程隊列鎖 
60. pthread_mutex_t _thread_mutex; 
61. //線程隊列 
62. list<threadInfo*> _threadList; 
63.  
64. ///初始化，創建最小個數線程/// 
65. void initThread();  
66. ///任務分配線程/// 
67. static void* taskAllocation(void*arg); 
68. pthread_t tasktPid; 
69. ///線程銷毀、狀態檢查線程/// 
70. static void* checkThread(void* arg); 
71. pthread_t checktPid; 
72. bool checkrun; 
73.  
74. //線程異常退出清理 
75. static void threadCleanUp(void* arg); 
76.  
77. // 
78. int addThread(list<threadInfo*> *plist,threadInfo* ptinfo); 
79.  
80. public: 
81. CthreadPoolManage(); 
82. /* 
83. 保留的最少線程，最多線程數，空閑多久銷毀，保留幾個線程的冗余 
84. */ 
85. CthreadPoolManage(int min,int max,int waitSec); 
86. ~CthreadPoolManage(); 
87.  
88. int start(); 
89. //任務注入器 
90. int putDuty(CDoit *,void *); 
91.  
92. int getNowThreadNum(); 
93.  
94. }; 
95.  
96. #endif // CTHREADPOOLMANAGE_H 

CPP

1. /* 
2. * 線程池，線程管理類 
3. *  
4. */ 
5.  
6. #include "cthreadpoolmanage.h" 
7.  
8. CthreadPoolManage::CthreadPoolManage() 
9. { 
10. _minThreads = 5; //最少保留幾個線程 
11. _maxThreads = 5; //最多可以有幾個線程 
12. _waitSec = 10; //空閑多少秒后將線程關閉 
13. pthread_mutex_init(&_duty_mutex, NULL); 
14. pthread_mutex_init(&_thread_mutex, NULL); 
15. checkrun = true; 
16. } 
17.  
18.  
19. CthreadPoolManage::CthreadPoolManage(int min, int max, int waitSec) 
20. { 
21. CthreadPoolManage(); 
22. _minThreads = min; //最少保留幾個線程 
23. _maxThreads = max; //最多可以有幾個線程 
24. _waitSec = waitSec; //空閑多少秒后將線程關閉 
25. } 
26.  
27. CthreadPoolManage::~CthreadPoolManage() 
28. { 
29.  
30. } 
31. void CthreadPoolManage::threadCleanUp(void* arg) 
32. { 
33. threadInfo* tinfo = (threadInfo*)arg; 
34. tinfo->isbusy = false; 
35. pthread_mutex_unlock(&tinfo->mtx); 
36. pthread_attr_destroy (&tinfo->cThreadAttr); 
37. delete tinfo; 
38. } 
39.  
40. void* CthreadPoolManage::startThread(void* arg) 
41. { 
42. cout<<"線程開始工作"<<endl; 
43. threadInfo* tinfo = (threadInfo*)arg; 
44. pthread_cleanup_push(threadCleanUp,arg); 
45. while(tinfo->doFlag){ 
46. pthread_mutex_lock(&tinfo->mtx); 
47. if(tinfo->doit == NULL) 
48. { 
49. cout<<"開始等待任務"<<endl; 
50. pthread_cond_wait(&tinfo->cond,&tinfo->mtx); 
51. cout<<"有任務了"<<endl; 
52. } 
53. tinfo->isbusy = true; 
54. tinfo->doit->start(tinfo->value); 
55. tinfo->doit->end(); 
56. tinfo->doit=NULL; 
57. tinfo->isbusy = false; 
58. time( &tinfo->beginTime); 
59. pthread_mutex_unlock(&tinfo->mtx); 
60. } 
61. //0正常執行到這兒不執行清理函數，異常會執行 
62. pthread_cleanup_pop(0); 
63. pthread_attr_destroy (&tinfo->cThreadAttr); 
64. delete tinfo; 
65. cout<<"線程結束"<<endl; 
66. } 
67.  
68. void CthreadPoolManage::initThread() 
69. { 
70. int i = 0; 
71. for(i = 0;i<this->_minThreads;i++) 
72. { 
73. threadInfo *tinfo = new threadInfo; 
74. tinfo->doit = NULL; 
75. tinfo->value = NULL; 
76. tinfo->isbusy = false; 
77. tinfo->doFlag = true; 
78. // PTHREAD_CREATE_DETACHED (分離線程) 和 PTHREAD _CREATE_JOINABLE (非分離線程) 
79. pthread_attr_init(&tinfo->cThreadAttr); 
80. pthread_attr_setdetachstate(&tinfo->cThreadAttr,PTHREAD_CREATE_DETACHED ); 
81. cout<<"初始化了一個線程"<<endl; 
82. if(pthread_create(&tinfo->cThreadPid,&tinfo->cThreadAttr,startThread,(void *)tinfo) != 0) 
83. { 
84. cout<<"創建線程失敗"<<endl; 
85. break; 
86. } 
87. this->_threadList.push_back(tinfo); 
88. } 
89. } 
90.  
91. int CthreadPoolManage::addThread(std::list< CthreadPoolManage::threadInfo* >* plist, CthreadPoolManage::threadInfo* ptinfo) 
92. { 
93. threadInfo *tinfo = new threadInfo; 
94. tinfo->doit = ptinfo->doit; 
95. tinfo->value = ptinfo->value; 
96. tinfo->isbusy = true; 
97. if(pthread_create(&tinfo->cThreadPid,NULL,startThread,(void *)tinfo) != 0) 
98. { 
99. cout<<"創建線程失敗"<<endl; 
100. return -1; 
101. } 
102. plist->push_back(tinfo); 
103. return 0; 
104. } 
105.  
106.  
107. int CthreadPoolManage::putDuty(CDoit* doit, void* value) 
108. { 
109. threadInfo *tinfo = new threadInfo; 
110. time( &tinfo->beginTime); 
111. tinfo->doit= doit; 
112. tinfo->value = value; 
113. pthread_mutex_lock(&_duty_mutex); 
114. this->_dutyList.push_back(tinfo); 
115. pthread_mutex_unlock(&_duty_mutex); 
116. return 0; 
117. } 
118.  
119. void* CthreadPoolManage::taskAllocation(void*arg) 
120. { 
121. CthreadPoolManage * ptmanage = (CthreadPoolManage*)arg; 
122. int size_1 = 0; 
123. int size_2 = 0; 
124. int i_1 = 0; 
125. int i_2 = 0; 
126. bool a_1 = true; 
127. bool a_2 = true; 
128. threadInfo* ptinfo; 
129. threadInfo* ptinfoTmp; 
130. while(true){ 
131. size_1 = 0; 
132. size_2 = 0; 
133. pthread_mutex_lock(&ptmanage->_duty_mutex); 
134. pthread_mutex_lock(&ptmanage->_thread_mutex); 
135. size_1 = ptmanage->_dutyList.size(); 
136. size_2 =ptmanage->_threadList.size(); 
137. for(list<threadInfo*>::iterator itorti1 = ptmanage->_dutyList.begin();itorti1 !=ptmanage->_dutyList.end();) 
138. { 
139. ptinfo = *itorti1; 
140. a_1 = true; 
141. for(list<threadInfo*>::iterator itorti2 = ptmanage->_threadList.begin();itorti2!=ptmanage->_threadList.end();itorti2++){ 
142. ptinfoTmp = *itorti2; 
143. if(EBUSY == pthread_mutex_trylock(&ptinfoTmp->mtx)) 
144. { 
145. continue; 
146. } 
147. if(!ptinfoTmp->isbusy) 
148. { 
149. ptinfoTmp->doit = ptinfo->doit; 
150. ptinfoTmp->value = ptinfo->value; 
151. ptinfoTmp->isbusy = true; 
152. pthread_cond_signal(&ptinfoTmp->cond); 
153. pthread_mutex_unlock(&ptinfoTmp->mtx); 
154. a_1 = false; 
155. delete ptinfo; 
156. break; 
157. } 
158. pthread_mutex_unlock(&ptinfoTmp->mtx); 
159. } 
160. if(a_1){ 
161. if(ptmanage->_threadList.size()>ptmanage->_maxThreads||ptmanage->addThread(&ptmanage->_threadList,ptinfo)!=0) 
162. { 
163. itorti1++; 
164. continue; 
165. }else{ 
166. itorti1 = ptmanage->_dutyList.erase(itorti1); 
167. } 
168. delete ptinfo; 
169. }else{ 
170. itorti1 = ptmanage->_dutyList.erase(itorti1); 
171. } 
172. } 
173. pthread_mutex_unlock(&ptmanage->_duty_mutex); 
174. pthread_mutex_unlock(&ptmanage->_thread_mutex); 
175. usleep(USLEEP_TIME); 
176. } 
177. return 0; 
178. } 
179.  
180. void* CthreadPoolManage::checkThread(void* arg) 
181. { 
182. CthreadPoolManage * ptmanage = (CthreadPoolManage*)arg; 
183. threadInfo* ptinfo; 
184. time_t nowtime; 
185. while(ptmanage->checkrun){ 
186. sleep(CHECK_TIME); 
187. pthread_mutex_lock(&ptmanage->_thread_mutex); 
188. if(ptmanage->_threadList.size()<=ptmanage->_minThreads) 
189. { 
190. pthread_mutex_unlock(&ptmanage->_thread_mutex); 
191. continue; 
192. } 
193. for(list<threadInfo*>::iterator itorti2 = ptmanage->_threadList.begin();itorti2!=ptmanage->_threadList.end();){ 
194. ptinfo = *itorti2; 
195. if(EBUSY == pthread_mutex_trylock(&ptinfo->mtx)) 
196. { 
197. itorti2++; 
198. continue; 
199. } 
200. time(&nowtime); 
201. if(ptinfo->isbusy == false && nowtime-ptinfo->beginTime>ptmanage->_waitSec) 
202. { 
203. ptinfo->doFlag = false; 
204. itorti2 = ptmanage->_threadList.erase(itorti2); 
205. }else{ 
206. itorti2++; 
207. } 
208. pthread_mutex_unlock(&ptinfo->mtx); 
209. } 
210. pthread_mutex_unlock(&ptmanage->_thread_mutex); 
211. } 
212. } 
213.  
214. int CthreadPoolManage::start() 
215. { 
216. //初始化 
217. this->initThread(); 
218. //啟動任務分配線程 
219. if(pthread_create(&tasktPid,NULL,taskAllocation,(void *)this) != 0) 
220. { 
221. cout<<"創建任務分配線程失敗"<<endl; 
222. return -1; 
223. } 
224. //創建現程狀態分配管理線程 
225. if(pthread_create(&checktPid,NULL,checkThread,(void *)this) != 0) 
226. { 
227. cout<<"創建線程狀態分配管理線程失敗"<<endl; 
228. return -1; 
229. } 
230. return 0; 
231. } 
232.  
233. /////////////////////////////// 
234. int CthreadPoolManage::getNowThreadNum() 
235. { 
236. int num = 0; 
237. pthread_mutex_lock(&this->_thread_mutex); 
238. num = this->_threadList.size(); 
239. pthread_mutex_unlock(&this->_thread_mutex); 
240. return num ; 
241. }