关于完成端口的问题，请各位高人们指点一下

kleavings

我在网上看到一些关于完成端口的例子，如下：
首先是 ServerHandler.h

2. #ifndef _SERVER_HANDER_H_
3. #define _SERVER_HANDER_H_
5. #pragma once
8. #include "ace/Proactor.h"
9. #include "ace/Asynch_IO.h"
10. #include "ace/message_block.h"
12. class ServerHander :public ACE_Service_Handler
13. {
14. public:
15.         ServerHander(void);
16.         virtual ~ServerHander(void);
17.         static void SetSleepTime(const DWORD t)
18.         {
19.                 m_sleepTime = t;
20.         };
22.         virtual void open(ACE_HANDLE h, ACE_Message_Block& _mb);
24. protected:
25.         virtual void handle_read_stream(const ACE_Asynch_Read_Stream::Result &result);
27.         virtual void handle_write_stream(const ACE_Asynch_Write_Stream::Result &result);
29.         virtual void  handle_time_out(const ACE_Time_Value &tv, const void *p);
31.         void DisplayInfo(ACE_HANDLE h, char* str) const;
34. private:
35.         ACE_Asynch_Read_Stream        m_reader;
36.         ACE_Asynch_Write_Stream m_writer;
38.         static DWORD                        m_sleepTime;
39.         time_t                                        m_lastTime;
40. };
43. #endif
45. ----------------------------------------------------------------------------------------
47. -------------------------------------------
48. ServerHander.cpp
50. #include "StdAfx.h"
51. #include "ServerHander.h"
52. #include "ace/OS_NS_sys_socket.h"
53. #include "ace/INET_Addr.h"
54. #include "ace/SOCK_SEQPACK_Association.h"
55. #include "ace/OS.h"
57. #define TIME_OUT 10
59. DWORD ServerHander::m_sleepTime = 0;
61. ServerHander::ServerHander(void):m_lastTime(0)
62. {
63. }
65. ServerHander::~ServerHander(void)
66. {
67.         //关闭
68.         if (this->handle() != ace_invalid_handle)               
69.         {
70.                 //显示客户端连接地址和端口
71.                 displayinfo(this->handle(), " disconnected.");
73.                 ace_proactor::instance()->cancel_timer(*this,1);
74.                 ace_os::shutdown(this->handle(), sd_both);
75.                 ace_os::closesocket( this->handle() );
76.                 this->handle(ace_invalid_handle);
77.         }
78. }
80. //客户端连接
81. void ServerHander::open(ACE_HANDLE h,
82.                                                 ACE_Message_Block& _mb)
83. {
84.         this->handle(h);
86.         //记录时间
87.         m_lastTime = ACE_OS::time(NULL);
89.         //ACE_Proactor::instance()->schedule_timer(*this, 0, ACE_Time_Value(0),
91. ACE_Time_Value(TIME_OUT));
93.         //构造I/O流
94.         if( this->m_reader.open(*this) != 0 || this->m_writer.open(*this) != 0 )
95.         {
96.                 cout<<"m_reader or m_writer open failed..."<<endl;
97.                 delete this;
98.                 return;
99.         }
101.         //显示客户端连接地址和端口        
102.         DisplayInfo(this->handle(), " connected.");
104.         ACE_Message_Block* mb = NULL;
105.         ACE_NEW_NORETURN(mb, ACE_Message_Block(1024));        
107.         //发起读操作
108.         if( this->m_reader.read( *mb, mb->space() ) != 0 )
109.         {
110.                 cout<<"m_reader read failed..."<<endl;
111.                 mb->release();
112.                 delete this;
113.         }
114. }
116. //读操作完成
117. void ServerHander::handle_read_stream(const ACE_Asynch_Read_Stream::Result &result)
118. {
119.         //记录时间
120.         m_lastTime = ACE_OS::time(NULL);
122.         ACE_Message_Block &mb = result.message_block();
124.         //传输不成功
125.         if ( (!result.success()) || (result.bytes_transferred() == 0) )
126.         {
127.                 cout<<"Read failed..."<<endl;
129.                 mb.release();
130.                 delete this;  
131.         }
132.         else        //接收完成
133.         {
134.                 //等待 模拟过载导致的响应速度变慢
135.                 //Sleep( m_sleepTime );
137.                /////////////////////////////////////////
138.                ///////////////////////////////////////
139.                //加载上自己所需要的类
141.               ///////////////////////////////////////////
142.               ////////////////////////////////////////
144.                 //写回
145.                 //mb.wr_ptr(0);
146.                 //mb.wr_ptr()[-2] = 0x03;
147.                 if (this->m_writer.write( mb, mb.length() ) == -1)
148.                 {
149.                         cout<<"Server write failed..."<<endl;
150.                         mb.release();
151.                 }
153.                
154.                 else        //写回成功，再继续读下一组数据
155.                 {
156.                         ACE_Message_Block *new_mb = NULL;
157.                         ACE_NEW_NORETURN(new_mb, ACE_Message_Block(1024));
159.                         this->m_reader.read(*new_mb, new_mb->space());
161.                         cout<<"Read again."<<endl;
162.                 }
163.         }
164. }
166. //写操作完成
167. void ServerHander::handle_write_stream(const ACE_Asynch_Write_Stream::Result &result)
168. {
169.         cout<<"Write completed."<<endl;
170.         //释放消息
171.         result.message_block().release();
172. }
174. //超时
175. void ServerHander::handle_time_out(const ACE_Time_Value &tv,
176.                                                                         const void *p)
177. {
178.         time_t curTime = ACE_OS::time(NULL);
180.         if( curTime - m_lastTime > TIME_OUT )
181.         {
182.                 cout<<"TimeOut"<<endl;
183.                 delete this;
184.         }
185. }
187. //显示信息
188. void ServerHander::DisplayInfo(ACE_HANDLE h,
189.                                                            char* str) const
190. {
191.         //获取客户端连接地址和端口
192.         ACE_INET_Addr addr;
193.         ACE_SOCK_SEQPACK_Association ass = ACE_SOCK_SEQPACK_Association(h);
194.         size_t addr_size=1;
195.         ass.get_remote_addr(addr);
197.         cout<< addr.get_host_addr() <<":"<< addr.get_port_number() <<str<<endl;
198. }

复制代码

[ 本帖最后由 winston 于 2007-12-19 12:20 编辑 ]
2007-12-19 11:30 rosebush
我将echoserver的代码贴出供大家研究测试2

多线程的proactor部分

2. proactorTask.h
3. #ifndef _CPROACTOR_TASK_H_
4. #define _CPROACTOR_TASK_H_
7. #pragma once
9. #include "ace\Task_T.h"
10. #include "ace\Thread_Semaphore.h"
11. #include "ace\Proactor.h"
12. #include "ace\WIN32_Proactor.h"
15. class CProactorTask :public ACE_Task<ACE_MT_SYNCH>
16. {
17. public:
18.         CProactorTask(void);
19.         virtual ~CProactorTask(void);
21.         int Start(const int nMax);
22.         int Stop(void);
23.         int Create(void);
24.         int Release(void);
25.         virtual int svc(void);
27. protected:
28.         ACE_Thread_Semaphore        m_sem;                                //信号量
29.         ACE_Proactor*                        m_pProactor;                //完成端口对象指
31. 针
32. };
35. #endif
36. ----------------------------------------------------------------------------------------
38. -----------
39. cpp部分
41. #include "StdAfx.h"
42. #include "ProactorTask.h"
44. CProactorTask::CProactorTask(void)
45. {
46. }
48. CProactorTask::~CProactorTask(void)
49. {
50. }
53. //
54. //创建完成端口对象
55. //
56. int CProactorTask::Create(void)
57. {
58.         ACE_WIN32_Proactor *proactor_impl = 0;
60.         //新建
61.         ACE_NEW_RETURN(proactor_impl, ACE_WIN32_Proactor, -1);
63.         //关联
64.         ACE_NEW_RETURN(this->m_pProactor, ACE_Proactor(proactor_impl, 1 ), -1);
66.         //保存
67.         ACE_Proactor::instance(this->m_pProactor, 1);
69.         return 0;
70. }
73. //
74. //启动线程池
75. //
76. int CProactorTask::Start(const int nMax)        //线程数量
77. {
78.         //创建完成端口对象
79.         Create();
81.         //创建线程
82.         this->activate(THR_NEW_LWP, nMax);
84.         int i;
85.         //保证所有线程已启动
86.         for(i = nMax; i>0; i--)
87.         {
88.                 m_sem.acquire();        //Block the thread until the semaphore count
90. becomes greater than 0, then decrement it.
91.         }
93.         cout<<"Start."<<endl;
94.         return 0;
95. }
98. //
99. //删除线程池
100. //
101. int CProactorTask::Stop(void)
102. {
103.         ACE_Proactor::end_event_loop();
104.         this->wait();
105.         return 0;
106. }
109. //
110. //每个线程调用
111. //
112. int CProactorTask::svc(void)
113. {
114.         ACE_DEBUG((LM_INFO,ACE_TEXT("svc函数调用!\n")));
116.         //Increment the semaphore by 1
117.         m_sem.release(1);
118.         ACE_Proactor::run_event_loop();
119.         return 0;
120. }
123. //
124. //释放
125. //
126. int CProactorTask::Release(void)
127. {
128.         ACE_Proactor::close_singleton();
129.         m_pProactor = 0;
131.         cout<<"Release."<<endl;
132.         return 0;
133. }
134. ----------------------------------------------------------------------------------------
136. ----------------
138. main.cpp部分
140. // EchoServer.cpp : Defines the entry point for the console application.
141. //
142. // 2007.12.6
143. // Echo Server Proactor模式
146. #include "Winbase.h"
147. #include "stdafx.h"
149. #include "ace/INET_Addr.h"
150. #include "ace/Asynch_Acceptor.h"
151. #include "ServerHander.h"
152. #include "ProactorTask.h"
154. int _tmain(int argc, _TCHAR* argv[])
155. {
156.         cout<<"******* Echo Server *******"<<endl<<endl;
158.         //获得CPU数量
159.         SYSTEM_INFO sysInfo;
160.         GetSystemInfo(&sysInfo);
161.         int threadNum = sysInfo.dwNumberOfProcessors<<1;        // CPU * 2
163.         //开启线程
164.         CProactorTask task;
165.         task.Start( threadNum );
167.         ACE_Asynch_Acceptor<ServerHander> MyAcceptor;
169.         ACE_INET_Addr addr(5050);
171.         if(MyAcceptor.open(addr) == -1)
172.         {
173.                 cout<<"acceptor open failed..."<<endl;
174.                 return 1;
175.         }
177.         cout<<"Listening on "<< addr.get_port_number() <<"..."<<endl;
179.         DWORD sleepTime = 0;
181.         while(1)
182.         {
183.                 cin>>sleepTime;
184.                 ServerHander::SetSleepTime(sleepTime);
185.                 cout<<"********** Set sleep time to "<<sleepTime<<" ************"<<endl;
186.         }
188.         return 0;
189. }

复制代码

遇到了以下几个问题，请高人们指点一下：
1 ， 在 "加载上自己所需要的类“ 的地方加载上我需要用的内容，是不是就可以达到并发的效果了

，还是在其他的地方加载呢？

2，当在 "加载上自己所需要的类" 地方写上个循环作为替代如下：
                for (int i=0;i<= 99999;i++)
                {
                        for (int j=0;j<= 50000;j++)
                        {
                               
                                if ( (i == 55500) && (j  == 1110))
                                {
                                        cout<<"la la la ..."<<endl;
                                }
                        }
                }
                cout<<"finish"<<endl;

当在服务器端启用多个线程时，当有多个客户端发出请求时：
我会发现 ：
当有第一个进入 handle_read_stream 后，就开始执行我的循环去了，其他的请求还有open里，而无

法得到响应，不是按照时间片原则，应该会得到响应吗？这是为什么呢？ 而我要在 我的循环前面加

个 sleep 等几个请求同时进入了 handle_read_stream 后，他们几个到是并行工作的，这又是为什么

呢？ 总之希望在什么时候都能达到并行工作的效果以达到并发，请高人指点一下~~~

3, 在 handle_time_out 里，如果超时就delete this他，因为不在同一个线程中，可能会造成崩溃，

有没有什么解决办法呢？

谢谢各位高人给我一些指点~~~~

winston

1、并发是Proactor提供的。ServerHander 只是数据的处理，不管并发。
2、写个死循环，占满了CPU，OS怎么给你切换？
3、这个问题是因为你同步工作没做好。做好同步，就不会出任何问题。

kleavings

那请问您我该把我所需要用的类放到哪里呢？
怎么做才能达到并发的效果呢？
能不能给个例子呢，谢谢~~~

kleavings

请高人们赐教一下~~~

modern

是这样的，在你标记的地方加载你的处理类，当然不是一个good idea啦，
如版主所说，你把CPU都占满了，完成端口上哪调度去。

一般的做法是，开一个task的线程池，在你标记的地方，把你从网络接受过来的
数据包扔到线程队列里去处理，在线程队列内部，加载上你自己实现的报文解析类，处理类等等
这样才可以达到并发的效果。

delete崩溃的问题肯定是你自己没有使用正确的加锁保护，
这个得根据实际情况具体分析了。

winston

modern 说了我想表达的一切，正解！

kleavings

谢谢高人指点，回都研究一下，再向您们讨教~~~

gsqgst

望大家多多指导