BOOST 线程完全攻略 - 扩展 - 事务线程

时间：2022-03-19 20:18:46
什么叫事务线程
举个例子：
我们写一个IM客户端的登录子线程，则该子线程会有这么几个事务要处理
No.1 TCP Socket物理连接
No.2 逻辑登录
No.3 好友在线查询
No.4 状态更新
我们通常的代码写法是
  1. void ThreadLogin()
  2. {
  3. try
  4. {
  5. if(fail(物理连接))
  6. throw;
  7. if(fail(登录))
  8. throw;
  9. if(fail(查询好友))
  10. throw;
  11. if(fail(更新))
  12. throw;
  13. }
  14. catch(exception)
  15. {
  16. }
  17. }

串行的逻辑用串行的代码写，不太好看，况且中途如果主线程发出取消指令，还不好处理。

这里扩展的thread类，就是来解决这个问题的，他会提供给程序员一种事件处理的模式
  1. class threadLogin
  2. {
  3. void onEventConnect()
  4. {
  5. 物理连接
  6. }
  7. void onEventLogin()
  8. {
  9. 登录
  10. }
  11. void onEventQuery()
  12. {
  13. 查询
  14. }
  15. void onEventUpdate()
  16. {
  17. 更新
  18. }
  19. }
源码如下
  1. // thread.hpp : controlled_module_ex类的扩展
  2. // 增强线程事务处理能力
  3. #pragma once
  4. #include "controlled_module_ex.hpp"
  5. class thread: public controlled_module_ex
  6. {
  7. protected:
  8. static const int NONE = -1;
  9. static const int WAITING =-2;
  10. static const int DONE =-3;
  11. static const int FAILED =-4;
  12. protected:
  13. struct process
  14. {
  15. int level;
  16. int status;
  17. int sequence;
  18. int trycount;
  19. int tryindex;
  20. std::string lasterror;
  21. double timeout;
  22. bool bTimeout;
  23. };
  24. process m_process;
  25. controlled_timer m_timer_process;
  26. int m_process_begin,m_process_end;
  27. double m_timeout_default;
  28. public:
  29. void startprocess(int process_begin,int process_end,double timeout_default=1.0,int cycle=1000)
  30. {
  31. m_process_begin = process_begin;
  32. m_process_end = process_end;
  33. m_timeout_default = timeout_default;
  34. m_process.level = m_process_begin;
  35. m_process.tryindex = 0;
  36. this->postmessage(BM_RING_PROCESS);
  37. m_timer_process.starttimer(cycle,this);
  38. }
  39. void tryagain()
  40. {
  41. if(this->m_process.level==thread::NONE)
  42. return;
  43. this->m_process.tryindex++;
  44. if(this->m_process.trycount>0 && this->m_process.tryindex>=this->m_process.trycount)
  45. {
  46. this->fail();
  47. }
  48. else
  49. this->postmessage(BM_RING_PROCESS);
  50. }
  51. void next()
  52. {
  53. if(this->m_process.level==thread::NONE)
  54. return;
  55. if(this->m_process.level>=this->m_process_end)
  56. {
  57. this->m_timer_process.stoptimer();
  58. this->postmessage(BM_RING_PROCESSEND);
  59. }
  60. else
  61. {
  62. this->m_process.tryindex = 0;
  63. this->m_process.level++;
  64. this->m_process.bTimeout = false;
  65. this->postmessage(BM_RING_PROCESS);
  66. }
  67. }
  68. void fail()
  69. {
  70. m_process.level = thread::NONE;
  71. this->m_timer_process.stoptimer();
  72. this->postmessage(BM_RING_PROCESSFAIL);
  73. }
  74. virtual void on_safestart()
  75. {
  76. m_process.level = thread::NONE;
  77. m_process.status = thread::NONE;
  78. m_process_begin = m_process_end = thread::NONE;
  79. controlled_module_ex::on_safestart();
  80. }
  81. virtual void on_safestop()
  82. {
  83. m_timer_process.stoptimer();
  84. controlled_module_ex::on_safestop();
  85. }
  86. virtual void message(const _command & cmd)
  87. {
  88. controlled_module_ex::message(cmd);
  89. if(cmd.nCmd==BM_RING_PROCESS)
  90. {
  91. this->on_process();
  92. }
  93. if(cmd.nCmd==BM_RING_PROCESSEND)
  94. {
  95. this->m_process.level = thread::NONE;
  96. this->on_process_end();
  97. }
  98. if(cmd.nCmd==BM_RING_PROCESSFAIL)
  99. {
  100. this->m_process.level = thread::NONE;
  101. this->on_process_fail();
  102. }
  103. }
  104. virtual void on_timer(const controlled_timer * p)
  105. {
  106. if(p==this->m_timer_process)
  107. {
  108. if(this->m_process.level!=thread::NONE)
  109. {
  110. if(this->m_process.level>=this->m_process_begin && this->m_process.level<=this->m_process_end)
  111. {
  112. if(this->m_process.status==thread::NONE)
  113. {
  114. this->m_process.level = this->m_process_begin;
  115. m_process.tryindex = 0;
  116. on_process();
  117. }
  118. else if(this->m_process.status==thread::WAITING)
  119. {
  120. if(this->m_process.timeout>0)
  121. {
  122. time_t cur;
  123. time(&cur);
  124. if(difftime(cur,(time_t)this->m_process.sequence)>this->m_process.timeout)
  125. {
  126. this->m_process.bTimeout = true;
  127. this->tryagain();
  128. }
  129. }
  130. }
  131. else if(this->m_process.status==thread::FAILED)
  132. {
  133. this->tryagain();
  134. }
  135. else if(this->m_process.status==thread::DONE)
  136. {
  137. this->m_process.level++;
  138. m_process.tryindex = 0;
  139. this->on_process();
  140. }
  141. }
  142. }
  143. }
  144. }
  145. virtual void on_process()
  146. {
  147. time((time_t*)&m_process.sequence);
  148. m_process.timeout = m_timeout_default;
  149. m_process.status = thread::WAITING;
  150. m_process.trycount = -1;
  151. }
  152. virtual void on_process_end(){}
  153. virtual void on_process_fail(){}
  154. int get_sequence(){return m_process.sequence;}
  155. void put_timeout(double v){m_process.timeout = v;}
  156. void put_trycount(int v){m_process.trycount = v;}
  157. int get_level(){return m_process.level;}
  158. void put_level(int v){m_process.level=v;}
  159. std::string get_lasterror(){return m_process.lasterror;}
  160. void put_lasterror(std::string v){m_process.lasterror=v;}
  161. __declspec(property(put=put_trycount)) int trycount;
  162. __declspec(property(put=put_timeout)) double timeout;
  163. __declspec(property(get=get_level,put=put_level)) int level;
  164. __declspec(property(get=get_sequence)) int sequence;
  165. __declspec(property(get=get_lasterror,put=put_lasterror)) std::string lasterror;
  166. };

虚拟函数thread::on_process()处理各种事务事件

虚拟函数thread::on_process_end()是所有事务处理完毕事件
虚拟函数thread::on_process_fail()是事务处理出现错误，这时所有事务被取消，线程终止
这里给一个简单的范例，
总共线程要完成3件事务，其中第二个事务要求用户确认是否继续
 
  1. #define PROCESS_1   1
  2. #define PROCESS_2   2
  3. #define PROCESS_3   3
  4. class thdex: public thread
  5. {
  6. public:
  7. virtual void on_process()
  8. {
  9. thread::on_process();
  10. if(this->level==PROCESS_1)
  11. {
  12. cout << "work on process 1..." << endl;
  13. Sleep(100);
  14. cout << "process 1 done." << endl;
  15. this->next();
  16. }
  17. else if(this->level==PROCESS_2)
  18. {
  19. cout << "work on process 2..." << endl;
  20. this->timeout = -1;
  21. if(IDNO==::MessageBox(0,"are your want continue?","ask",MB_ICONQUESTION|MB_YESNO))
  22. {
  23. this->lasterror = "canceled by user";
  24. this->fail();
  25. }
  26. else
  27. {
  28. Sleep(100);
  29. cout << "process 2 done." << endl;
  30. this->next();
  31. }
  32. }
  33. else if(this->level==PROCESS_3)
  34. {
  35. cout << "work on process 3..." << endl;
  36. Sleep(100);
  37. cout << "process 3 done." << endl;
  38. this->next();
  39. }
  40. }
  41. virtual void on_process_fail()
  42. {
  43. cout << this->lasterror << endl;
  44. }
  45. virtual void on_process_end()
  46. {
  47. cout << "all process done." << endl;
  48. }
  49. };
  50. int _tmain(int argc, _TCHAR* argv[])
  51. {
  52. thdex t;
  53. t.safestart();
  54. t.startprocess(PROCESS_1,PROCESS_3);
  55. char buf[10];
  56. gets_s(buf,sizeof buf);
  57. t.safestop();
  58. return 0;
  59. }

相关文章