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

1. // controlled_module_ex.hpp : controlled_module类的扩展
2. // 增强线程之间消息通讯
3. // 增加线程安全启动和安全关闭功能
4. // 增加定时器功能
5. 
   
6. #pragma once
7. #include 
8. #include 
9. #include "controlled_module.hpp"
10. 
    
11. struct _command
12. {
13.     typedef boost::shared_ptr<_command> CCmdPtr;
14.     unsigned int nCmd;
15.     boost::any anyParam;
16. };
17. struct _wait_command
18. {       
19.     boost::any par;
20.     unsigned int command;
21.     void * event;
22.     boost::shared_ptr resp;
23. };
24. class controlled_module_ex;
25. struct _notify
26. {
27.     controlled_module_ex * sender;
28.     int id;
29.     boost::any par;
30. };
31. #define BM_RESERVE 1000
32. #define BM_RING_START BM_RESERVE+1
33. #define BM_RING_STOP BM_RESERVE+2
34. #define BM_RING_SETTIME  BM_RESERVE+3
35. #define BM_RING_SETPARENT BM_RESERVE+4
36. #define BM_RING_CYCLE BM_RESERVE+5
37. #define BM_RING_PROCESS BM_RESERVE+6
38. #define BM_RING_PROCESSEND BM_RESERVE+7
39. #define BM_RING_PROCESSFAIL BM_RESERVE+8
40. #define BM_TIMER    BM_RESERVE+9
41. #define BM_COMMAND  BM_RESERVE+10
42. #define BM_NOTIFY   BM_RESERVE+11
43. 
    
44. #define BM_USER 9000
45. class controlled_timer;
46. class controlled_module_ex: public controlled_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<typename T>
59.     bool postmessage(unsigned int nCmd, const boost::shared_ptr& p)
60.     {
61.         if(this==0||!m_safe)return false;
62.         boost::mutex::scoped_lock lock(m_mutex_command);
63.         _command::CCmdPtr cmd(new _command);
64.         cmd->nCmd = nCmd;
65.         cmd->anyParam = p;
66.         m_list_command.push_back(cmd);
67.         return true;
68.     }
69.     boost::any execute(unsigned int command,boost::any par,int timeout=-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(new boost::any);
77.         if(this->postmessage(BM_COMMAND,shared))
78.         {
79.             DWORD dw = WaitForSingleObject(cmd.event,timeout);
80.             CloseHandle(cmd.event);
81.             if(dw!=WAIT_OBJECT_0)
82.                 return boost::any();
83.             else
84.                 return *cmd.resp;
85.         }
86.         else
87.         {
88.             CloseHandle(cmd.event);
89.             return boost::any();
90.         }
91.     }
92.     void notify(_notify p)
93.     {
94.         this->postmessage(BM_NOTIFY,p);
95.     }
96.     bool postmessage(unsigned int nCmd,boost::any p)
97.     {
98.         if(this==0||!m_safe)
99.             return false;
100.         boost::mutex::scoped_lock lock(m_mutex_command);
101.         _command::CCmdPtr cmd(new _command);
102.         cmd->nCmd = nCmd;
103.         cmd->anyParam = p;
104.         m_list_command.push_back(cmd);
105.         return true;
106.     }
107.     bool postmessage(unsigned int nCmd)
108.     {
109.         if(this==0||!m_safe)
110.             return false;
111.         boost::mutex::scoped_lock lock(m_mutex_command);
112.         _command::CCmdPtr cmd(new _command);
113.         cmd->nCmd = nCmd;
114.         cmd->anyParam = 0;
115.         m_list_command.push_back(cmd);
116.         return true;
117.     }
118.     virtual bool work()
119.     {
120.         if(!getmessage())
121.             return false;
122.         else
123.         {
124.             Sleep(this->m_sleeptime);
125.             return true;
126.         }
127.     }
128.     virtual void message(const _command & cmd)
129.     {
130.         if(cmd.nCmd==BM_RING_START)
131.         {
132.             this->on_safestart();
133.         }
134.         else if(cmd.nCmd==BM_RING_STOP)
135.         {
136.             this->on_safestop();
137.         }
138.         else if(cmd.nCmd==BM_TIMER)
139.         {
140.             this->on_timer(boost::any_cast(cmd.anyParam));
141.         }
142.         else if(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.         else if(cmd.nCmd==BM_NOTIFY)
150.         {
151.             try
152.             {
153.                 _notify par = boost::any_cast<_notify>(cmd.anyParam);
154.                 this->on_notify(par);
155.             }
156.             catch(boost::bad_any_cast)
157.             {
158.             }
159.         }
160.     }
161.     virtual void release()
162.     {
163.         boost::mutex::scoped_lock lock(m_mutex_command);
164.         m_list_command.clear();
165.     }
166.     void safestart()
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.     void safestop()
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_lock lock(m_mutex_command);
184.             _command::CCmdPtr cmd(new _command);
185.             cmd->nCmd = BM_RING_STOP;
186.             cmd->anyParam = 0;
187.             m_list_command.push_back(cmd);
188.         }
189.         DWORD dw = ::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.     virtual void on_timer(const controlled_timer * p){}
198.     virtual void on_safestart()
199.     {
200.         SetEvent(m_safestart_event);
201.     }
202.     virtual void on_safestop()
203.     {
204.         SetEvent(m_safestop_event);
205.     }
206.     virtual void on_notify(const _notify & p)
207.     {
208.     }
209. protected:
210.     virtual boost::any on_command(const unsigned int command,const boost::any par)
211.     {
212.         return boost::any();
213.     }
214.     bool getmessage()
215.     {
216.         std::list<_command::CCmdPtr> cache;
217.         {
218.             boost::mutex::scoped_lock lock(m_mutex_command);
219.             while(!m_list_command.empty())
220.             {
221.                 _command::CCmdPtr p = m_list_command.front();
222.                 m_list_command.pop_front();
223.                 cache.push_back(p);
224.             }
225.         }
226.         _command::CCmdPtr stop_command;
227.         std::list<_command::CCmdPtr>::iterator item;
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::CCmdPtr p = 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.             return true;
256.         }
257.         else
258.         {
259.             cache.clear();
260.             this->message(*stop_command);
261.             return false;
262.         }
263.     }
264. private:
265.     void*   m_safestart_event;
266.     void* m_safestop_event;
267.     bool m_safe;//在多线程，尤其牵涉到线程之间有类似socket级别关联时，当父线程safestop以后有可能会收到其他线程的postmessage，这时会引起线程死锁，这个m_safe就是解决这个问题的，当safestop以后不再接收新消息处理
268.     boost::mutex m_mutex_command;
269.     std::list<_command::CCmdPtr> m_list_command;
270. };
271. class controlled_timer: public controlled_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.   int m_time;
285.   int m_step;
286. public:
287.   void starttimer(int time,controlled_module_ex* parent)
288.   {
289.     this->safestart();
290.     this->postmessage(BM_RING_SETPARENT,parent);
291.     this->postmessage(BM_RING_SETTIME,time);
292.   }
293.   void stoptimer()
294.   {
295.     this->safestop();
296.   }
297. public:
298.   virtual void on_safestop()
299.   {
300.     m_time = 0;
301.     controlled_module_ex::on_safestop();
302.   }
303.   virtual void message(const _command & cmd)
304.   {
305.     controlled_module_ex::message(cmd);
306.     if(cmd.nCmd==BM_RING_SETTIME)
307.     {
308.         int time = boost::any_cast<int>(cmd.anyParam);
309.         this->m_time = time/this->m_sleeptime;
310.         this->postmessage(BM_RING_CYCLE);
311.     }
312.     else if(cmd.nCmd==BM_RING_SETPARENT)
313.     {
314.         this->m_parent  = boost::any_cast(cmd.anyParam);
315.     }   
316.     else if(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"
2. 
   
3. 
   
4. class thdex: public controlled_module_ex
5. {
6. protected:
7.     virtual void message(const _command & cmd)
8.     {
9.         controlled_module_ex::message(cmd);
10.         if(cmd.nCmd==BM_USER+1)
11.         {
12.             cout << "get message" << endl;
13.         }
14.     }
15. };
16. 
    
17. int _tmain(int argc, _TCHAR* argv[])
18. {
19.     thdex t;
20.     t.safestart();
21.     t.postmessage(BM_USER+1);
22.     char buf[10];
23.     gets_s(buf,sizeof buf);
24.     t.safestop();
25.     return 0;
26. }

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

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

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

1. #include "controlled_module_ex.hpp"
2. 
   
3. struct mystruct
4. {
5.     string a;
6.     int b;
7. };
8. class thdex: public controlled_module_ex
9. {
10. protected:
11.     virtual void message(const _command & cmd)
12.     {
13.         controlled_module_ex::message(cmd);
14.         if(cmd.nCmd==BM_USER+1)
15.         {
16.             cout << "get integer:" << boost::any_cast<int>(cmd.anyParam) << endl;
17.         }
18.         if(cmd.nCmd==BM_USER+2)
19.         {
20.             cout << "get string:" << boost::any_cast(cmd.anyParam) << endl;
21.         }
22.         if(cmd.nCmd==BM_USER+3)
23.         {
24.             mystruct par = boost::any_cast(cmd.anyParam);
25.             cout << "get mystruct:" << par.a << "," << par.b << endl;
26.         }
27.     }
28. };
29. 
    
30. int _tmain(int argc, _TCHAR* argv[])
31. {
32.     thdex t;
33.     t.safestart();
34.     t.postmessage(BM_USER+1,123);
35.     t.postmessage(BM_USER+2,string("hello world"));
36.     mystruct par;
37.     par.a = "hello world";
38.     par.b = 123;
39.     t.postmessage(BM_USER+3,par);
40. 
    
41.     char buf[10];
42.     gets_s(buf,sizeof buf);
43.     t.safestop();
44.     return 0;
45. }
46. 
    
47. 
    

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

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

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

1. #include "controlled_module_ex.hpp"
2. 
   
3. struct mystruct
4. {
5.     boost::shared_ptr<char> data;
6.     int datalen;
7. };
8. class thdex: public controlled_module_ex
9. {
10. protected:
11.     virtual void message(const _command & cmd)
12.     {
13.         controlled_module_ex::message(cmd);
14.         if(cmd.nCmd==BM_USER+1)
15.         {
16.             cout << "get sharedptr" << endl; //仅仅得到数据，得不到数据长度
17.         }
18.         if(cmd.nCmd==BM_USER+2)
19.         {
20.             mystruct par = boost::any_cast(cmd.anyParam);
21.             cout << "get sharedptr len:" << par.datalen << endl;
22.         }
23.     }
24. };
25. 
    
26. int _tmain(int argc, _TCHAR* argv[])
27. {
28.     thdex t;
29.     t.safestart();
30.     t.postmessage(BM_USER+1,boost::shared_ptr<char>(new char[1000]));
31.     mystruct par;
32.     par.datalen = 1000;
33.     par.data = boost::shared_ptr<char>(new char[par.datalen]);
34.     t.postmessage(BM_USER+2,par);
35. 
    
36.     char buf[10];
37.     gets_s(buf,sizeof buf);
38.     t.safestop();
39.     return 0;
40. }
41. 
    
42. 
    

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"
2. 
   
3. class thdex: public controlled_module_ex
4. {
5. protected:
6.     boost::any on_command(const unsigned int command,const boost::any par)
7.     {
8.         if(command==1)
9.         {
10.             cout << "on command" << endl;
11.             return 0;
12.         }
13.         if(command==2)
14.         {
15.             cout << "on command,par:" << boost::any_cast(par) << endl;
16.             return 0;
17.         }
18.         if(command==3)
19.         {
20.             return true;
21.         }
22.         else
23.             return controlled_module_ex::on_command(command,par);
24.     }
25. };
26. 
    
27. int _tmain(int argc, _TCHAR* argv[])
28. {
29.     thdex t;
30.     t.safestart();
31.     t.execute(1,0);//等待子线程处理完成
32.     t.execute(2,string("hello world"));//带参数 等待子线程完成
33.     bool rs = boost::any_cast<bool>(t.execute(3,0));//等待子线程处理完成，并取得返回值
34.     cout << "get thread result:" << rs << endl;
35.     boost::any timeout = t.execute(4,0,1000);//等待子线程处理，超时1秒
36.     if(timeout.empty())
37.         cout << "timeout " << endl;
38. 
    
39.     char buf[10];
40.     gets_s(buf,sizeof buf);
41.     t.safestop();
42.     return 0;
43. }

4.定时器

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

1. #include "controlled_module_ex.hpp"
2. 
   
3. class thdex: public controlled_module_ex
4. {
5. protected:
6.     virtual void on_timer(const controlled_timer *p)
7.     {
8.         cout << "ontimer" << endl;
9.     }
10. };
11. 
    
12. int _tmain(int argc, _TCHAR* argv[])
13. {
14.     thdex t;
15.     controlled_timer timer;
16.     t.safestart();
17.     timer.starttimer(1000,&t);
18. 
    
19.     char buf[10];
20.     gets_s(buf,sizeof buf);
21.     t.safestop();
22.     return 0;
23. }