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

2023-03-19 22:03:10

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

1.向线程PostMessage

  函数controlled_module_ex::postmessage完成消息推送。
  虚拟函数controlled_module_ex::message(const _command & cmd)实现消息接收。
  1. #include "controlled_module_ex.hpp"
  2. class thdex: public controlled_module_ex
  3. {
  4. protected:
  5. virtual void message(const _command & cmd)
  6. {
  7. controlled_module_ex::message(cmd);
  8. if(cmd.nCmd==BM_USER+1)
  9. {
  10. cout << "get message" << endl;
  11. }
  12. }
  13. };
  14. int _tmain(int argc, _TCHAR* argv[])
  15. {
  16. thdex t;
  17. t.safestart();
  18. t.postmessage(BM_USER+1);
  19. char buf[10];
  20. gets_s(buf,sizeof buf);
  21. t.safestop();
  22. return 0;
  23. }
  2.向线程PostMessage，并携带简单对象参数
 我们都知道常规的PostMessage要传参，如果是整型参数，还可以用强制转换，但如果是其他类型，例如字符串，我们就必须创建一个字符串缓冲，把缓冲指针作为参数传过去，线程还不能忘记删除，否则导致内存泄漏，自定义结构也是一样的操作，如果想尝试传递一个CString对象，是不可能的。
  幸运的是boost提供了boost::any来抽象任何对象类型，controlled_module_ex的消息传递都是基于boost::any来完成，程序员可以由此写出干净而且内存安全的代码。
  1. #include "controlled_module_ex.hpp"
  2. struct mystruct
  3. {
  4. string a;
  5. int b;
  6. };
  7. class thdex: public controlled_module_ex
  8. {
  9. protected:
  10. virtual void message(const _command & cmd)
  11. {
  12. controlled_module_ex::message(cmd);
  13. if(cmd.nCmd==BM_USER+1)
  14. {
  15. cout << "get integer:" << boost::any_cast<int>(cmd.anyParam) << endl;
  16. }
  17. if(cmd.nCmd==BM_USER+2)
  18. {
  19. cout << "get string:" << boost::any_cast<string>(cmd.anyParam) << endl;
  20. }
  21. if(cmd.nCmd==BM_USER+3)
  22. {
  23. mystruct par = boost::any_cast<mystruct>(cmd.anyParam);
  24. cout << "get mystruct:" << par.a << "," << par.b << endl;
  25. }
  26. }
  27. };
  28. int _tmain(int argc, _TCHAR* argv[])
  29. {
  30. thdex t;
  31. t.safestart();
  32. t.postmessage(BM_USER+1,123);
  33. t.postmessage(BM_USER+2,string("hello world"));
  34. mystruct par;
  35. par.a = "hello world";
  36. par.b = 123;
  37. t.postmessage(BM_USER+3,par);
  38. char buf[10];
  39. gets_s(buf,sizeof buf);
  40. t.safestop();
  41. return 0;
  42. }

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

  假如我们书写了一个录音子线程，如何将录制的语音数据传递给其他线程呢，常规做法是创建一个缓冲，将语音数据填充进去，然后将缓冲地址作为参数传递，这种做法要求接收线程不能忘记删除，否则会导致内存泄漏。
  幸运的是boost提供了智能指针，可以类似java,c#的智能内存回收一样来管理内存分配，我们如果使用这个对象来作为参数传递，就可以完美的防范内存泄漏行为，就算子线程没有处理，别担心，内存它会自动回收的。
  1. #include "controlled_module_ex.hpp"
  2. struct mystruct
  3. {
  4. boost::shared_ptr<char> data;
  5. int datalen;
  6. };
  7. class thdex: public controlled_module_ex
  8. {
  9. protected:
  10. virtual void message(const _command & cmd)
  11. {
  12. controlled_module_ex::message(cmd);
  13. if(cmd.nCmd==BM_USER+1)
  14. {
  15. cout << "get sharedptr" << endl; //仅仅得到数据，得不到数据长度
  16. }
  17. if(cmd.nCmd==BM_USER+2)
  18. {
  19. mystruct par = boost::any_cast<mystruct>(cmd.anyParam);
  20. cout << "get sharedptr len:" << par.datalen << endl;
  21. }
  22. }
  23. };
  24. int _tmain(int argc, _TCHAR* argv[])
  25. {
  26. thdex t;
  27. t.safestart();
  28. t.postmessage(BM_USER+1,boost::shared_ptr<char>(new char[1000]));
  29. mystruct par;
  30. par.datalen = 1000;
  31. par.data = boost::shared_ptr<char>(new char[par.datalen]);
  32. t.postmessage(BM_USER+2,par);
  33. char buf[10];
  34. gets_s(buf,sizeof buf);
  35. t.safestop();
  36. return 0;
  37. }

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

4.定时器

  类似于CWnd::OnTimer，controlled_module_ex也提供一个虚拟函数virtual void on_timer(const controlled_timer * p);来处理定时
  1. #include "controlled_module_ex.hpp"
  2. class thdex: public controlled_module_ex
  3. {
  4. protected:
  5. virtual void on_timer(const controlled_timer *p)
  6. {
  7. cout << "ontimer" << endl;
  8. }
  9. };
  10. int _tmain(int argc, _TCHAR* argv[])
  11. {
  12. thdex t;
  13. controlled_timer timer;
  14. t.safestart();
  15. timer.starttimer(1000,&t);
  16. char buf[10];
  17. gets_s(buf,sizeof buf);
  18. t.safestop();
  19. return 0;
  20. }
上一篇：vue快速入门


下一篇：在Ubuntu16.04.4上安装jdk