BOOST 线程完全攻略 - 扩展 - 线程消息通讯

转自：http://blog.csdn.net/IamNieo/archive/2008/09/10/2909236.aspx

1. //controlled_module_ex.hpp:controlled_module类的扩展
2. //增强线程之间消息通讯
3. //增加线程安全启动和安全关闭功能
4. //增加定时器功能
6. #pragmaonce
7. #include<boost/shared_ptr.hpp>
8. #include<boost/any.hpp>
9. #include"controlled_module.hpp"
11. struct_command
12. {
13. typedefboost::shared_ptr<_command>CCmdPtr;
14. unsignedintnCmd;
15. boost::anyanyParam;
16. };
17. struct_wait_command
18. {
19. boost::anypar;
20. unsignedintcommand;
21. void*event;
22. boost::shared_ptr<boost::any>resp;
23. };
24. classcontrolled_module_ex;
25. struct_notify
26. {
27. controlled_module_ex*sender;
28. intid;
29. boost::anypar;
30. };
31. #defineBM_RESERVE1000
32. #defineBM_RING_STARTBM_RESERVE+1
33. #defineBM_RING_STOPBM_RESERVE+2
34. #defineBM_RING_SETTIMEBM_RESERVE+3
35. #defineBM_RING_SETPARENTBM_RESERVE+4
36. #defineBM_RING_CYCLEBM_RESERVE+5
37. #defineBM_RING_PROCESSBM_RESERVE+6
38. #defineBM_RING_PROCESSENDBM_RESERVE+7
39. #defineBM_RING_PROCESSFAILBM_RESERVE+8
40. #defineBM_TIMERBM_RESERVE+9
41. #defineBM_COMMANDBM_RESERVE+10
42. #defineBM_NOTIFYBM_RESERVE+11
44. #defineBM_USER9000
45. classcontrolled_timer;
46. classcontrolled_module_ex:publiccontrolled_module
47. {
48. public:
49. controlled_module_ex()
50. {
51. m_safe=false;
52. }
53. ~controlled_module_ex()
54. {
55. safestop();
56. }
57. public:
58. template<typenameT>
59. boolpostmessage(unsignedintnCmd,constboost::shared_ptr<T>&p)
60. {
61. if(this==0||!m_safe)returnfalse;
62. boost::mutex::scoped_locklock(m_mutex_command);
63. _command::CCmdPtrcmd(new_command);
64. cmd->nCmd=nCmd;
65. cmd->anyParam=p;
66. m_list_command.push_back(cmd);
67. returntrue;
68. }
69. boost::anyexecute(unsignedintcommand,boost::anypar,inttimeout=-1)
70. {
71. boost::shared_ptr<_wait_command>shared(new_wait_command);
72. _wait_command&cmd=*shared;
73. cmd.command=command;
74. cmd.event=(void*)CreateEvent(0,FALSE,FALSE,0);
75. cmd.par=par;
76. cmd.resp=boost::shared_ptr<boost::any>(newboost::any);
77. if(this->postmessage(BM_COMMAND,shared))
78. {
79. DWORDdw=WaitForSingleObject(cmd.event,timeout);
80. CloseHandle(cmd.event);
81. if(dw!=WAIT_OBJECT_0)
82. returnboost::any();
83. else
84. return*cmd.resp;
85. }
86. else
87. {
88. CloseHandle(cmd.event);
89. returnboost::any();
90. }
91. }
92. voidnotify(_notifyp)
93. {
94. this->postmessage(BM_NOTIFY,p);
95. }
96. boolpostmessage(unsignedintnCmd,boost::anyp)
97. {
98. if(this==0||!m_safe)
99. returnfalse;
100. boost::mutex::scoped_locklock(m_mutex_command);
101. _command::CCmdPtrcmd(new_command);
102. cmd->nCmd=nCmd;
103. cmd->anyParam=p;
104. m_list_command.push_back(cmd);
105. returntrue;
106. }
107. boolpostmessage(unsignedintnCmd)
108. {
109. if(this==0||!m_safe)
110. returnfalse;
111. boost::mutex::scoped_locklock(m_mutex_command);
112. _command::CCmdPtrcmd(new_command);
113. cmd->nCmd=nCmd;
114. cmd->anyParam=0;
115. m_list_command.push_back(cmd);
116. returntrue;
117. }
118. virtualboolwork()
119. {
120. if(!getmessage())
121. returnfalse;
122. else
123. {
124. Sleep(this->m_sleeptime);
125. returntrue;
126. }
127. }
128. virtualvoidmessage(const_command&cmd)
129. {
130. if(cmd.nCmd==BM_RING_START)
131. {
132. this->on_safestart();
133. }
134. elseif(cmd.nCmd==BM_RING_STOP)
135. {
136. this->on_safestop();
137. }
138. elseif(cmd.nCmd==BM_TIMER)
139. {
140. this->on_timer(boost::any_cast<controlled_timer*>(cmd.anyParam));
141. }
142. elseif(cmd.nCmd==BM_COMMAND)
143. {
144. boost::shared_ptr<_wait_command>shared=boost::any_cast<boost::shared_ptr<_wait_command>>(cmd.anyParam);
145. _wait_command&cmd=*shared;
146. *cmd.resp=this->on_command(cmd.command,cmd.par);
147. SetEvent((HANDLE)cmd.event);
148. }
149. elseif(cmd.nCmd==BM_NOTIFY)
150. {
151. try
152. {
153. _notifypar=boost::any_cast<_notify>(cmd.anyParam);
154. this->on_notify(par);
155. }
156. catch(boost::bad_any_cast)
157. {
158. }
159. }
160. }
161. virtualvoidrelease()
162. {
163. boost::mutex::scoped_locklock(m_mutex_command);
164. m_list_command.clear();
165. }
166. voidsafestart()
167. {
168. if(!islive())
169. start();
170. m_safe=true;
171. m_safestart_event=(void*)CreateEvent(NULL,FALSE,FALSE,0);
172. postmessage(BM_RING_START);
173. ::WaitForSingleObject((HANDLE)m_safestart_event,INFINITE);
174. CloseHandle(m_safestart_event);
175. }
176. voidsafestop()
177. {
178. if(this->islive())
179. {
180. m_safe=false;
181. m_safestop_event=(void*)CreateEvent(NULL,FALSE,FALSE,0);
182. {
183. boost::mutex::scoped_locklock(m_mutex_command);
184. _command::CCmdPtrcmd(new_command);
185. cmd->nCmd=BM_RING_STOP;
186. cmd->anyParam=0;
187. m_list_command.push_back(cmd);
188. }
189. DWORDdw=::WaitForSingleObject((HANDLE)m_safestop_event,3*1000);
190. if(WAIT_OBJECT_0!=dw)
191. {
192. }
193. CloseHandle(m_safestop_event);
194. stop();
195. }
196. }
197. virtualvoidon_timer(constcontrolled_timer*p){}
198. virtualvoidon_safestart()
199. {
200. SetEvent(m_safestart_event);
201. }
202. virtualvoidon_safestop()
203. {
204. SetEvent(m_safestop_event);
205. }
206. virtualvoidon_notify(const_notify&p)
207. {
208. }
209. protected:
210. virtualboost::anyon_command(constunsignedintcommand,constboost::anypar)
211. {
212. returnboost::any();
213. }
214. boolgetmessage()
215. {
216. std::list<_command::CCmdPtr>cache;
217. {
218. boost::mutex::scoped_locklock(m_mutex_command);
219. while(!m_list_command.empty())
220. {
221. _command::CCmdPtrp=m_list_command.front();
222. m_list_command.pop_front();
223. cache.push_back(p);
224. }
225. }
226. _command::CCmdPtrstop_command;
227. std::list<_command::CCmdPtr>::iteratoritem;
228. for(item=cache.begin();item!=cache.end();item++)
229. {
230. if((*(*item)).nCmd==BM_RING_STOP)
231. {
232. stop_command=*item;
233. break;
234. }
235. }
236. if(stop_command.get()==0)
237. {
238. while(!cache.empty())
239. {
240. _command::CCmdPtrp=cache.front();
241. cache.pop_front();
242. try
243. {
244. if((*p).nCmd!=BM_RING_START)
245. {
246. if(!this->m_safe)
247. continue;
248. }
249. this->message(*p);
250. }
251. catch(boost::bad_any_cast&)
252. {
253. }
254. }
255. returntrue;
256. }
257. else
258. {
259. cache.clear();
260. this->message(*stop_command);
261. returnfalse;
262. }
263. }
264. private:
265. void*m_safestart_event;
266. void*m_safestop_event;
267. boolm_safe;//在多线程，尤其牵涉到线程之间有类似socket级别关联时，当父线程safestop以后有可能会收到其他线程的postmessage，这时会引起线程死锁，这个m_safe就是解决这个问题的，当safestop以后不再接收新消息处理
268. boost::mutexm_mutex_command;
269. std::list<_command::CCmdPtr>m_list_command;
270. };
271. classcontrolled_timer:publiccontrolled_module_ex
272. {
273. public:
274. controlled_timer()
275. {
276. this->m_time=0;
277. this->m_parent=0;
278. this->m_step=0;
279. }
280. ~controlled_timer(){
281. }
282. protected:
283. controlled_module_ex*m_parent;
284. intm_time;
285. intm_step;
286. public:
287. voidstarttimer(inttime,controlled_module_ex*parent)
288. {
289. this->safestart();
290. this->postmessage(BM_RING_SETPARENT,parent);
291. this->postmessage(BM_RING_SETTIME,time);
292. }
293. voidstoptimer()
294. {
295. this->safestop();
296. }
297. public:
298. virtualvoidon_safestop()
299. {
300. m_time=0;
301. controlled_module_ex::on_safestop();
302. }
303. virtualvoidmessage(const_command&cmd)
304. {
305. controlled_module_ex::message(cmd);
306. if(cmd.nCmd==BM_RING_SETTIME)
307. {
308. inttime=boost::any_cast<int>(cmd.anyParam);
309. this->m_time=time/this->m_sleeptime;
310. this->postmessage(BM_RING_CYCLE);
311. }
312. elseif(cmd.nCmd==BM_RING_SETPARENT)
313. {
314. this->m_parent=boost::any_cast<controlled_module_ex*>(cmd.anyParam);
315. }
316. elseif(cmd.nCmd==BM_RING_CYCLE)
317. {
318. if(m_time>0)
319. {
320. if(m_step>m_time)
321. {
322. m_parent->postmessage(BM_TIMER,this);
323. m_step=0;
324. }
325. m_step++;
326. }
327. this->postmessage(BM_RING_CYCLE);
328. }
329. }
330. };

1.向线程PostMessage

函数controlled_module_ex::postmessage完成消息推送。

虚拟函数controlled_module_ex::message(const _command & cmd)实现消息接收。

1. #include"controlled_module_ex.hpp"
4. classthdex:publiccontrolled_module_ex
5. {
6. protected:
7. virtualvoidmessage(const_command&cmd)
8. {
9. controlled_module_ex::message(cmd);
10. if(cmd.nCmd==BM_USER+1)
11. {
12. cout<<"getmessage"<<endl;
13. }
14. }
15. };
17. int_tmain(intargc,_TCHAR*argv[])
18. {
19. thdext;
20. t.safestart();
21. t.postmessage(BM_USER+1);
22. charbuf[10];
23. gets_s(buf,sizeofbuf);
24. t.safestop();
25. return0;
26. }

2.向线程PostMessage，并携带简单对象参数

我们都知道常规的PostMessage要传参，如果是整型参数，还可以用强制转换，但如果是其他类型，例如字符串，我们就必须创建一个字符串缓冲，把缓冲指针作为参数传过去，线程还不能忘记删除，否则导致内存泄漏，自定义结构也是一样的操作，如果想尝试传递一个CString对象，是不可能的。

幸运的是boost提供了boost::any来抽象任何对象类型，controlled_module_ex的消息传递都是基于boost::any来完成，程序员可以由此写出干净而且内存安全的代码。

1. #include"controlled_module_ex.hpp"
3. structmystruct
4. {
5. stringa;
6. intb;
7. };
8. classthdex:publiccontrolled_module_ex
9. {
10. protected:
11. virtualvoidmessage(const_command&cmd)
12. {
13. controlled_module_ex::message(cmd);
14. if(cmd.nCmd==BM_USER+1)
15. {
16. cout<<"getinteger:"<<boost::any_cast<int>(cmd.anyParam)<<endl;
17. }
18. if(cmd.nCmd==BM_USER+2)
19. {
20. cout<<"getstring:"<<boost::any_cast<string>(cmd.anyParam)<<endl;
21. }
22. if(cmd.nCmd==BM_USER+3)
23. {
24. mystructpar=boost::any_cast<mystruct>(cmd.anyParam);
25. cout<<"getmystruct:"<<par.a<<","<<par.b<<endl;
26. }
27. }
28. };
30. int_tmain(intargc,_TCHAR*argv[])
31. {
32. thdext;
33. t.safestart();
34. t.postmessage(BM_USER+1,123);
35. t.postmessage(BM_USER+2,string("helloworld"));
36. mystructpar;
37. par.a="helloworld";
38. par.b=123;
39. t.postmessage(BM_USER+3,par);
41. charbuf[10];
42. gets_s(buf,sizeofbuf);
43. t.safestop();
44. return0;
45. }

3.向线程PostMessage,并传递内存块参数

假如我们书写了一个录音子线程，如何将录制的语音数据传递给其他线程呢，常规做法是创建一个缓冲，将语音数据填充进去，然后将缓冲地址作为参数传递，这种做法要求接收线程不能忘记删除，否则会导致内存泄漏。

幸运的是boost提供了智能指针，可以类似java,c#的智能内存回收一样来管理内存分配，我们如果使用这个对象来作为参数传递，就可以完美的防范内存泄漏行为，就算子线程没有处理，别担心，内存它会自动回收的。

1. #include"controlled_module_ex.hpp"
3. structmystruct
4. {
5. boost::shared_ptr<char>data;
6. intdatalen;
7. };
8. classthdex:publiccontrolled_module_ex
9. {
10. protected:
11. virtualvoidmessage(const_command&cmd)
12. {
13. controlled_module_ex::message(cmd);
14. if(cmd.nCmd==BM_USER+1)
15. {
16. cout<<"getsharedptr"<<endl;//仅仅得到数据，得不到数据长度
17. }
18. if(cmd.nCmd==BM_USER+2)
19. {
20. mystructpar=boost::any_cast<mystruct>(cmd.anyParam);
21. cout<<"getsharedptrlen:"<<par.datalen<<endl;
22. }
23. }
24. };
26. int_tmain(intargc,_TCHAR*argv[])
27. {
28. thdext;
29. t.safestart();
30. t.postmessage(BM_USER+1,boost::shared_ptr<char>(newchar[1000]));
31. mystructpar;
32. par.datalen=1000;
33. par.data=boost::shared_ptr<char>(newchar[par.datalen]);
34. t.postmessage(BM_USER+2,par);
36. charbuf[10];
37. gets_s(buf,sizeofbuf);
38. t.safestop();
39. return0;
40. }

3.向线程SendMessage

函数controlled_module_ex::execute完成这个工作

虚拟函数controlled_module_ex::on_command(const unsigned int command,const boost::any par)响应消息处理

1. #include"controlled_module_ex.hpp"
3. classthdex:publiccontrolled_module_ex
4. {
5. protected:
6. boost::anyon_command(constunsignedintcommand,constboost::anypar)
7. {
8. if(command==1)
9. {
10. cout<<"oncommand"<<endl;
11. return0;
12. }
13. if(command==2)
14. {
15. cout<<"oncommand,par:"<<boost::any_cast<string>(par)<<endl;
16. return0;
17. }
18. if(command==3)
19. {
20. returntrue;
21. }
22. else
23. returncontrolled_module_ex::on_command(command,par);
24. }
25. };
27. int_tmain(intargc,_TCHAR*argv[])
28. {
29. thdext;
30. t.safestart();
31. t.execute(1,0);//等待子线程处理完成
32. t.execute(2,string("helloworld"));//带参数等待子线程完成
33. boolrs=boost::any_cast<bool>(t.execute(3,0));//等待子线程处理完成，并取得返回值
34. cout<<"getthreadresult:"<<rs<<endl;
35. boost::anytimeout=t.execute(4,0,1000);//等待子线程处理，超时1秒
36. if(timeout.empty())
37. cout<<"timeout"<<endl;
39. charbuf[10];
40. gets_s(buf,sizeofbuf);
41. t.safestop();
42. return0;
43. }

4.定时器

类似于CWnd::OnTimer，controlled_module_ex也提供一个虚拟函数virtual void on_timer(const controlled_timer * p);来处理定时

1. #include"controlled_module_ex.hpp"
3. classthdex:publiccontrolled_module_ex
4. {
5. protected:
6. virtualvoidon_timer(constcontrolled_timer*p)
7. {
8. cout<<"ontimer"<<endl;
9. }
10. };
12. int_tmain(intargc,_TCHAR*argv[])
13. {
14. thdext;
15. controlled_timertimer;
16. t.safestart();
17. timer.starttimer(1000,&t);
19. charbuf[10];
20. gets_s(buf,sizeofbuf);
21. t.safestop();
22. return0;
23. }