初学C++，发个线程池

shtr

小弟才学C++不久，为练手写的线程池，还望各位大牛不吝赐教，多多拍砖。
threadpool.h:

1. /**
   
2. * @file    threadpool.h
   
3. * @brief    本文件用来实现一个不依赖第三方库与具体操作系统的线程池
   
4. *
   
5. * detail...
   
6. *
   
7. * @author    shtr
   
8. * @version    1.0
   
9. * @date    2009年12月6日
   
10. *
    
11. * @see        
    
12. *
    
13. * @par 版本记录：
    
14. * <table border=1>
    
15. *  <tr> <th>版本    <th>日期            <th>作者    <th>备注 </tr>
    
16. *  <tr> <td>1.0    <td>2009年12月6日    <td>shtr    <td>创建 </tr>
    
17. * </table>
    
18. */
    
19. #ifndef    _THREAD_POOL_
    
20. #define    _THREAD_POOL_
    
21. #ifdef WIN32
    
22. #    include <windows.h>
    
23. #    include <process.h>
    
24. #else
    
25. #    include <pthread.h>
    
26. #    include <semaphore.h>
    
27. #endif /* WIN32 */
    
28. 
    
29. #include <vector>
    
30. #include <queue>
    
31. 
    
32. 
    
33. class lock_i
    
34. {
    
35. public:
    
36.     virtual bool lock(void) = 0;
    
37.     virtual void unlock(void) = 0;
    
38. };
    
39. 
    
40. /**
    
41. * @class    class mutex_lock
    
42. *
    
43. * @brief    使用单值信号量封装的互斥锁
    
44. *
    
45. * detail...
    
46. *
    
47. * @author    shtr
    
48. * @date    2009年12月6日
    
49. *
    
50. * @see        
    
51. *
    
52. * @par 备注：
    
53. *
    
54. */
    
55. class mutex_lock
    
56.     : public lock_i
    
57. {
    
58. public:
    
59.     mutex_lock(void)
    
60.     {
    
61. #ifdef WIN32
    
62.         lock_ = CreateSemaphore(0, 1, 1, 0);
    
63. 
    
64. #else
    
65.         sem_init(&lock_, 0, 1);
    
66. #endif
    
67.     }
    
68.     ~mutex_lock(void)
    
69.     {
    
70. #ifdef WIN32
    
71.         if ( lock_ )
    
72.             CloseHandle(lock_), lock_ = 0;
    
73. #else
    
74.         sem_destroy(&lock_), lock_ = 0;
    
75. #endif
    
76.     }
    
77. 
    
78.     void unlock(void)
    
79.     {
    
80. #ifdef WIN32
    
81.         ReleaseSemaphore(lock_, 1, 0);
    
82. #else
    
83.         sem_post(&lock_);
    
84. #endif
    
85.     }
    
86. 
    
87.     bool lock(void)
    
88.     {
    
89. #ifdef WIN32
    
90.         DWORD ret = WaitForSingleObject(lock_, INFINITE);
    
91.         return ret == WAIT_OBJECT_0;
    
92. #else
    
93.         return sem_wait(&lock_) == 0;
    
94. #endif /* WIN32 */
    
95.     }
    
96. private:
    
97. #ifdef WIN32
    
98.     HANDLE                lock_;
    
99. #else
    
100.     sem_t               lock_;
     
101. #endif /* WIN32 */
     
102. };
     
103. 
     
104. /**
     
105. * @class    class scope_lock
     
106. *
     
107. * @brief    域锁
     
108. *
     
109. * 构造时请求锁，析构时释放锁
     
110. *
     
111. * @author    shtr
     
112. * @date    2009年12月6日
     
113. *
     
114. * @see        
     
115. *
     
116. * @par 备注：
     
117. *
     
118. */
     
119. class scope_lock
     
120.     : public lock_i
     
121. {
     
122. private:
     
123.     lock_i& lock_;
     
124. 
     
125. public:
     
126.     scope_lock(lock_i& lock)
     
127.         : lock_(lock)
     
128.     {
     
129. 
     
130.         lock_.lock();
     
131.     }
     
132.     ~scope_lock()
     
133.     {
     
134.         lock_.unlock();
     
135.     }
     
136. 
     
137.     bool lock(void)
     
138.     {
     
139.         return lock_.lock();
     
140.     }
     
141. 
     
142.     void unlock(void)
     
143.     {
     
144.         return lock_.unlock();
     
145.     }
     
146. };
     
147. 
     
148. /**
     
149. * @class    class thr_sem
     
150. *
     
151. * @brief    同步信号量
     
152. *
     
153. * detail...
     
154. *
     
155. * @author    shtr
     
156. * @date    2009年12月6日
     
157. *
     
158. * @see        
     
159. *
     
160. * @par 备注：
     
161. *
     
162. */
     
163. class thr_sem
     
164. {
     
165. public:
     
166.     thr_sem(void)
     
167.     {
     
168. #ifdef WIN32
     
169.         sem_ = CreateSemaphore(0, 0, INT_MAX, 0);
     
170. #else
     
171.         sem_init(&sem_, 0, 0);
     
172. #endif
     
173.     }
     
174.     ~thr_sem(void)
     
175.     {
     
176. #ifdef WIN32
     
177.         if ( sem_ )
     
178.             CloseHandle(sem_), sem_ = 0;
     
179. #else
     
180.         sem_destroy(&sem_), sem_ = 0;
     
181. #endif
     
182.     }
     
183. 
     
184.     void notify_one(void)
     
185.     {
     
186. #ifdef WIN32
     
187.         ReleaseSemaphore(sem_, 1, 0);
     
188. #else
     
189.         sem_post(&sem_);
     
190. #endif /* WIN32 */
     
191.     }
     
192. 
     
193.     bool timed_wait(int msec)
     
194.     {
     
195. #ifdef WIN32
     
196.         DWORD ret = WaitForSingleObject(sem_, msec);
     
197.         return ret == WAIT_OBJECT_0;
     
198. #else
     
199.         timespec tv;
     
200.         tv.tv_sec = time(NULL) + msec / 1000;
     
201.         tv.tv_nsec = 0;
     
202.         return sem_timedwait(&sem_, &tv) == 0;
     
203. #endif /* WIN32 */
     
204.     }
     
205. 
     
206.     bool wait(void)
     
207.     {
     
208. #ifdef WIN32
     
209.         DWORD ret = WaitForSingleObject(sem_, INFINITE);
     
210.         return ret == WAIT_OBJECT_0;
     
211. #else
     
212.         int ret = sem_wait(&sem_);
     
213.         if ( ret != 0)
     
214.             return errno;
     
215.         return ret;
     
216. #endif
     
217.     }
     
218. 
     
219. private:
     
220. #ifdef WIN32
     
221.     HANDLE                sem_;
     
222. #else
     
223.     sem_t               sem_;
     
224. #endif /* WIN32 */
     
225. };
     
226. 
     
227. 
     
228. class thr_ic
     
229. {
     
230. public:
     
231.     virtual void svc(void) = 0;
     
232. };
     
233. 
     
234. /**
     
235. * @class    class thr_impl
     
236. *
     
237. * @brief    线程实现接口，操作系统相关
     
238. *
     
239. * detail...
     
240. *
     
241. * @author    shtr
     
242. * @date    2009年12月6日
     
243. *
     
244. * @see        
     
245. *
     
246. * @par 备注：
     
247. *
     
248. */
     
249. class thr_impl
     
250. {
     
251. 
     
252. public:
     
253.     virtual int start(thr_ic* thr_unit)
     
254.     {
     
255.         int thr_id = create_thread(thr_unit);
     
256. 
     
257.         return thr_id;
     
258.     }
     
259. 
     
260.     virtual unsigned int create_thread(thr_ic* thr_unit) = 0;
     
261. 
     
262.     virtual int join(unsigned int tid) = 0;
     
263. 
     
264. };
     
265. #ifdef WIN32
     
266. class thr_impl_win : public thr_impl
     
267. {
     
268. public:
     
269.     virtual unsigned int create_thread(thr_ic* thr_unit)
     
270.     {
     
271.         return(unsigned int)_beginthread(svc, 0, thr_unit);
     
272.     }
     
273.     virtual int join(unsigned int tid)
     
274.     {
     
275.         HANDLE hThread = (HANDLE) tid;
     
276.         return WaitForSingleObject( hThread, INFINITE );
     
277.     }
     
278. private:
     
279.     static void svc(void* args)
     
280.     {
     
281.         thr_ic* unit = (thr_ic *)args;
     
282.         unit->svc();
     
283.     }
     
284. 
     
285. };
     
286. 
     
287. typedef thr_impl_win thr_impl_os;
     
288. #else
     
289. class thr_impl_nix : public thr_impl
     
290. {
     
291. public:
     
292.     virtual unsigned int create_thread(thr_ic* thr_unit)
     
293.     {
     
294.         pthread_t tid;
     
295.         int ret = pthread_create(&tid, NULL, svc, thr_unit);
     
296.         if (ret != 0)
     
297.             return errno;
     
298.         return tid;
     
299.     }
     
300.     virtual int join(unsigned int tid)
     
301.     {
     
302.         int ret = pthread_join(tid, NULL);
     
303.         if (ret != 0)
     
304.             return errno;
     
305.         return ret;
     
306.     }
     
307. private:
     
308.     static void* svc(void* args)
     
309.     {
     
310.         thr_ic* unit = (thr_ic *)args;
     
311.         unit->svc();
     
312.         return NULL;
     
313.     }
     
314. 
     
315. };
     
316. 
     
317. typedef thr_impl_nix thr_impl_os;
     
318. 
     
319. #endif /* WIN32 */
     
320. 
     
321. 
     
322. 
     
323. /**
     
324. * @class    class threadpool
     
325. *
     
326. * @brief    线程池管理类
     
327. *
     
328. * detail...
     
329. *
     
330. * @author    shtr
     
331. * @date    2009年12月6日
     
332. *
     
333. * @see        
     
334. *
     
335. * @par 备注：
     
336. *
     
337. */
     
338. template<class thr_impl = thr_impl_os, class sem_t = thr_sem>
     
339. class threadpool
     
340.     : public thr_ic
     
341. {
     
342. 
     
343. public:
     
344.     typedef void (*TaskFunc)(void *);
     
345.     typedef void* Arg;
     
346.     class Task
     
347.     {
     
348.     public:
     
349.         Task(TaskFunc func, Arg args)
     
350.             : func_(func)
     
351.             , args_(args)
     
352.         {    }
     
353.     public:
     
354.         TaskFunc    func_;
     
355.         Arg            args_;
     
356.     };
     
357.     typedef std::queue<Task> TaskQueue;
     
358. 
     
359.     TaskQueue        tasks_;            // 任务队列
     
360.     bool            thr_stop_;        // 终止标识
     
361. 
     
362. public:
     
363.     /* 工作线程 */
     
364.     class worker
     
365.         : public thr_ic
     
366.     {
     
367.     private:
     
368.         threadpool* pool_;
     
369.     public:
     
370.         worker(threadpool* pool = NULL)
     
371.             : pool_(pool)
     
372.         {
     
373.         }
     
374.         void bind_pool(threadpool* pool)
     
375.         {
     
376.             pool_ = pool;
     
377.         }
     
378. 
     
379.         virtual void svc(void)
     
380.         {
     
381.             while (!pool_->thr_stop_)
     
382.             {
     
383. 
     
384.                 if (!pool_->sem_.timed_wait(200))
     
385.                     continue;
     
386. 
     
387.                 pool_->lock_.lock();
     
388.                 Task task = pool_->tasks_.front();
     
389.                 pool_->tasks_.pop();
     
390.                 pool_->lock_.unlock();
     
391. 
     
392.                 (*task.func_)(task.args_);
     
393.             }
     
394.         }
     
395.     };
     
396. 
     
397.     worker                worker_;    // 工作线程
     
398. 
     
399. public:
     
400.     threadpool()
     
401.         : thrs_count_(0)
     
402.         , thrs_id_(NULL)
     
403.         , thr_stop_(false)
     
404.     {    }
     
405. 
     
406.     virtual void add_task(Task task)
     
407.     {
     
408. 
     
409.         lock_.lock();
     
410.         tasks_.push(task);
     
411.         lock_.unlock();
     
412.         sem_.notify_one();
     
413.     }
     
414. 
     
415.     virtual int open(unsigned int thrs_count)
     
416.     {
     
417.         // 创建线程
     
418.         thrs_count_ = thrs_count;
     
419. 
     
420.         worker_.bind_pool(this);
     
421.         this->start();
     
422. 
     
423.         return 0;
     
424. 
     
425.     }
     
426. 
     
427.     void start(void)
     
428.     {
     
429. 
     
430.         for (unsigned int i = 0; i < thrs_count_; i ++)
     
431.         {
     
432.             int thr_id = thr_impl_.start(&worker_);
     
433. 
     
434.             if ( thr_id > 0 )
     
435.             {
     
436.                 thrs_id_.push_back(thr_id);
     
437.             }
     
438.         }
     
439.     }
     
440. 
     
441.     void stop(void)
     
442.     {
     
443.         thr_stop_ = true;
     
444. 
     
445.         join_all();
     
446.     }
     
447. 
     
448. 
     
449.     void svc(void)
     
450.     {    }
     
451. 
     
452. private:
     
453. 
     
454.     void join_all(void)
     
455.     {
     
456.         for (std::vector<int>::const_iterator cit = thrs_id_.begin(); cit != thrs_id_.end(); cit++)
     
457.         {
     
458.             thr_impl_.join(*cit);
     
459.         }
     
460.     }
     
461. 
     
462. private:
     
463.     sem_t                sem_;            // 同步信号量
     
464.     mutex_lock            lock_;            // 互斥锁
     
465.     thr_impl            thr_impl_;        // 线程
     
466.     unsigned int        thrs_count_;    // 线程计数
     
467.     std::vector<int>    thrs_id_;
     
468. 
     
469. };
     
470. 
     
471. #endif /* _THREAD_POOL_ */

复制代码

测试：

1. #include "threadpool.h"
   
2. 
   
3. void func(void* args)
   
4. {
   
5.         printf("I'm %s.\n", (char *)args);
   
6. }
   
7. 
   
8. int _tmain(int argc, _TCHAR* argv[])
   
9. {
   
10.        
    
11.         threadpool<> pool;
    
12.    
    
13.         pool.add_task(threadpool<>::Task(func, "1"));
    
14.         pool.add_task(threadpool<>::Task(func, "2"));
    
15.         pool.add_task(threadpool<>::Task(func, "3"));
    
16.         pool.add_task(threadpool<>::Task(func, "4"));
    
17.         pool.add_task(threadpool<>::Task(func, "5"));
    
18.         pool.add_task(threadpool<>::Task(func, "6"));
    
19.         pool.add_task(threadpool<>::Task(func, "7"));
    
20.        
    
21.     pool.open(20);
    
22. 
    
23.         Sleep(2000);
    
24. 
    
25.         pool.add_task(threadpool<>::Task(func, "40"));
    
26.         pool.add_task(threadpool<>::Task(func, "50"));
    
27.         pool.add_task(threadpool<>::Task(func, "60"));
    
28.         pool.add_task(threadpool<>::Task(func, "70"));
    
29.        
    
30.         Sleep(2000);
    
31. 
    
32.         pool.stop();
    
33. 
    
34.        
    
35. 
    
36.         return 0;
    
37. }

复制代码

shtr

楼主有java的编程风格
c/unix 发表于 2010-05-12 11:52

    哪里？

liu1061

来支持一下,学习一下!!

yu_single

给楼主一点建议：
threadpool::Task这个类不够通用（原因是：typedef void (*TaskFunc)(void *)），这样仅支持C风格的带一个参数的函数，而把C++中的仿函数以及类成员函数排除在外了，这无疑是很大的一个硬伤，如需改进，参考《C++设计新思维》第5章“泛化仿函数”。