运用ACE_Thread_Manager类创建线程

winston

     创建线程需要要解决两个问题，一是调用线程函数，二是提供一个途径让线程能够访问到外部传递过来的参数。下面的代码演示了基本的用法：

1. #include
2. <stdexcept>
3. #include "ace/ACE.h"
4. #include
5. "ace/Log_Msg.h"
6. #include "ace/Thread_Manager.h"
7. #include
8. <map>
9. #include <string>
10. #include <iostream>
11. using
12. namespace std;
14. class ThreadArg {
15. public:
17.     ThreadArg()
18. {
19.     }
20. private:
21.     string arg0;
22. public:
24.     void
25. setArg0(string value) {
26.         arg0 = value;
27.     }
29.     string
30. getArg0() const {
31.         return arg0;
32.     }
33. };
35. class MyThread
36. {
37. public:
38.     static void* run_svc(void* arg) {
39.         
40. ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) MyThread running/n")));
41.         
42. ThreadArg* pArg = static_cast<ThreadArg*> (arg);
43.         cout <<
44. pArg->getArg0() << endl;
45.     }
46. };
48. int ACE_TMAIN(int,
49. ACE_TCHAR *[]) {
50.     ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Main Thread
51. running/n")));
52.     ThreadArg threadArg;
53.    
54. threadArg.setArg0("ok");
55.     ACE_thread_t threadID;
56.     if
57. (ACE_Thread_Manager::instance()->spawn(MyThread::run_svc,
58. static_cast<void*>(&threadArg),
59.             THR_DETACHED |
60. THR_SCOPE_SYSTEM, &threadID) == -1) {
61.         throw
62. std::runtime_error("can't create a new thread in SnapShotReqWiater::run
63. method");
64.     }
65.     ACE_Thread_Manager::instance()->wait();
66.    
67. return 0;
68. }

复制代码

     使用ACE_Thread_Manager::spawn方法创建线程，第一个参数是线程的函数，第二个是一个对象指针，里面存放了参数，其他的参数请参考文档
http://www.dre.vanderbilt.edu/Doxygen/5.7.5/html/ace/a00676.html#a36262a470e556182f5d69c4de7cfeaa1
    wait方法等待线程运行完毕后才会返回。

运用ACE_Task_Base类创建线程    前面一种方法不够面向对象，线程需要成为一个对象，并且参数可以通过设置属性的形式自然的进行。下面的例子来自于<<ACE Programmers Guide>>,略作修改：

1. #include "ace/Task.h"
2. #include
3. <string>
4. #include <iostream>
5. using namespace std;
7. class
8. MyThread : public ACE_Task_Base {
9. public:
11.     virtual int svc(void)
12. {
13.         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Handler Thread
14. running/n")));
15.         cout<<arg0<<endl;
16.         
17. ACE_OS::sleep(4);
18.         return 0;
19.     }
21.     void setArg0(string
22. const& arg0) {
23.         this->arg0 = arg0;
24.     }
26.     string
27. getArg0() const {
28.         return arg0;
29.     }
31. private:
32.     string
33. arg0;
34. };
36. int ACE_TMAIN(int, ACE_TCHAR *[]) {
37.    
38. ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Main Thread running/n")));
40.    
41. MyThread thread;
42.     string arg="ok";
43.     thread.setArg0(arg);
44.     int
45. result = thread.activate();
46.     ACE_ASSERT(result == 0);
48.    
49. thread.wait();
50.     return 0;
51. }

复制代码

    MyThread类可以添加一些属性，svc成员函数可以内部访问这些属性。activate内部的实现代码实际上用的还是ACE_Thread_Manager::spawn_n方法。现在我们很清楚，通过继承ACE_Task_Base类来创建线程是更方便和优雅的。

ACE_Task
   线程之间常常需要通信，ACE有一种机制叫做消息队列。要获得这种能力，我们只需要使用ACE_Task_Base的子类ACT_Task即可。上面的例 子代码进行了一些修改，MyThread类从ACE_Task<>模板类派生，并且内部有一个循环，每隔四秒钟就会检查一下有没有消息，如果 消息类型是ACE_Message_Block::MB_STOP，则结束线程。

1. #include "ace/Task_T.h"
2. #include
3. <string>
4. #include <iostream>
5. using namespace std;
7. class
8. MyThread : public ACE_Task<ACE_MT_SYNCH> {
9. public:
11.     virtual
12. int svc(void) {
13.         ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Handler Thread
14. running/n")));
15.         cout << arg0 << endl;
16.         for (;;)
17. {
18.             ACE_OS::sleep(4);
19.             ACE_Message_Block *
20. pMsg;
21.             ACE_Time_Value time(0, 100);
22.             if (getq(pMsg,
23. &time) != -1) {
24.                 if (pMsg->msg_type() ==
25. ACE_Message_Block::MB_STOP) {
26.                     
27. pMsg->release();
28.                     
29. msg_queue_->close();
30.                     
31. ACE_Thread::exit(0);
32.                 }
33.             }
34.         
35. }
36.         return 0;
37.     }
39.     void setArg0(string const& arg0)
40. {
41.         this->arg0 = arg0;
42.     }
44.     string getArg0() const
45. {
46.         return arg0;
47.     }
49. private:
50.     string
51. arg0;
52. };
54. int ACE_TMAIN(int, ACE_TCHAR *[]) {
55.    
56. ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Main Thread running/n")));
58.    
59. MyThread thread;
60.     string arg="ok";
61.     thread.setArg0(arg);
62.     int
63. result = thread.activate();
64.     ACE_ASSERT(result == 0);
66.    
67. ACE_Message_Block* pMessage=new ACE_Message_Block(1);
68.    
69. pMessage->msg_type(ACE_Message_Block::MB_STOP);
70.    
71. thread.putq(pMessage);
72.    
73.     thread.wait();
74.     return
75. 0;
76. }

复制代码

注意：
1）putq和getq的第二个参数都是代表超时的时间，缺省为一直等待。
2）ACE_Message_Block作为消息，一般在发送者处用new创建，在接受者处调用release()方法销毁
3）msg_queue_->close();是关闭消息队列，使得发送者无法再发送消息。
4）msg_queue_->is_full()可以查看消息队列是否已经满
5) msg_queue_->high_water_mark()返回消息队列可以容纳的最后一个消息的索引，默认从0开始计数。我的机器上是默认最大索引是16384，也就是可以存放16385个消息。
6) msg_queue_->high_water_mark(size_t size)可以让我们重新设置消息队列的最大索引。

    关于ACE_Task的消息队列和Windows消息队列的比较，可以看这篇文章：http://blog.csdn.net/imjj/archive/2006/08/19/1097248.aspx
最后给一个比较复杂的消息队列的应用，利用ACE_OutputCDR类将对象序列化到ACE_Message_Block中，然后加入消息队列，另一个线程取出后，通过ACE_InputCDR类反序列化回对象。

1. /*
3. * File:   main.cpp
4. * Author: chenshu
5. *
6. * Created on December 1,
7. 2009, 7:21 PM
8. */
11. #include "ace/Task_T.h"
12. #include
13. <string>
14. #include <iostream>
15. using namespace std;
16. #include
17. "ace/CDR_Stream.h"
19. class FileMessage {
20. public:
21.     static const
22. ACE_CDR::ULong createdFile=0;
24. public:
25.     ACE_CString getFolderPath()
26. const {
27.         return folderPath_;
28.     }
30.     void
31. setFolderPath(ACE_CString const& path) {
32.         folderPath_ =
33. path;
34.     }
36.     ACE_CString getFileName() const {
37.         return
38. fileName_;
39.     }
41.     void setFileName(ACE_CString const& name)
42. {
43.         fileName_ = name;
44.     }
46.     ACE_CDR::ULong
47. getFileMessageType() const{
48.         return messageType_;
49.     }
51.    
52. void setFileMessageType(ACE_CDR::ULong type) {
53.         messageType_ =
54. type;
55.     }
57.     ACE_CDR::ULong getSize() const {
58.         return
59. sizeof(ACE_CDR::ULong)+fileName_.length()+1+
60.                
61. sizeof(ACE_CDR::ULong)+
62.                
63. sizeof(ACE_CDR::ULong)+folderPath_.length()+1;
64.     }
65. private:
66.    
67. ACE_CString fileName_;
68.     ACE_CDR::ULong messageType_;
69.     ACE_CString
70. folderPath_;
71. };
73. int operator<<(ACE_OutputCDR & cdr,
74. FileMessage const& message) {
75.    
76. cdr<<message.getFileName();
77.    
78. cdr<<message.getFileMessageType();
79.    
80. cdr<<message.getFolderPath();
81.     return
82. cdr.good_bit();
83. }
85. int operator>>(ACE_InputCDR & cdr,
86. FileMessage & message) {
87.     ACE_CString name;
88.    
89. cdr>>name;
90.     message.setFileName(name);
91.     ACE_CDR::ULong
92. type;
93.     cdr >> type;
94.     message.setFileMessageType(type);
95.    
96. ACE_CString folderPath;
97.     cdr>>folderPath;
98.    
99. message.setFolderPath(folderPath);
100.     return
101. cdr.good_bit();
102. }
104. class MyThread : public ACE_Task<ACE_MT_SYNCH>
105. {
106. public:
108.     virtual int svc(void) {
109.         ACE_DEBUG((LM_DEBUG,
110. ACE_TEXT("(%t) Handler Thread running/n")));
111.         cout << arg0
112. << endl;
113.         cout << "high_water_mark:" <<
114. msg_queue_->high_water_mark() << endl;
115.         for (;;)
116. {
117.             ACE_OS::sleep(4);
118.             ACE_Message_Block *
119. pMsg;
120.             ACE_Time_Value time(0, 100);
121.             if (getq(pMsg,
122. &time) != -1) {
123.                 if (pMsg->msg_type() ==
124. ACE_Message_Block::MB_STOP) {
125.                     
126. pMsg->release();
127.                     
128. msg_queue_->close();
129.                     
130. ACE_Thread::exit(0);
131.                 } else if(ACE_Message_Block::MB_DATA)
132. {
133.                     ACE_InputCDR inputCDR(pMsg);
134.                     
135. FileMessage message;
136.                     
137. inputCDR>>message;
138.                 }
139.             }
140.         
141. }
142.         return 0;
143.     }
145.     void setArg0(string const& arg0)
146. {
147.         this->arg0 = arg0;
148.     }
150.     string getArg0() const
151. {
152.         return arg0;
153.     }
155. private:
156.     string
157. arg0;
158. };
160. int ACE_TMAIN(int, ACE_TCHAR *[]) {
161.    
162. ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%t) Main Thread running/n")));
164.    
165. MyThread thread;
166.     string arg = "ok";
167.     thread.setArg0(arg);
168.    
169. int result = thread.activate();
170.     ACE_ASSERT(result == 0);
172.    
173. FileMessage message;
174.    
175. message.setFileMessageType(FileMessage::createdFile);
176.    
177. message.setFileName("chenshu");
178.    
179. message.setFolderPath("/home/chenshu");
180.     ACE_OutputCDR
181. outputCDR(message.getSize()+ACE_CDR::MAX_ALIGNMENT);
182.    
183. outputCDR<<message;
184.     ACE_Message_Block*
185. pMessage=const_cast<ACE_Message_Block*>(outputCDR.begin());
186.    
187. thread.putq(pMessage);
189.     thread.wait();
190.     return 0;
191. }

复制代码

http://blog.csdn.net/wallwind/article/details/7220083