Posted on 2007-06-19 16:31
小明 閱讀(7656)
評論(1) 編輯 收藏 引用 所屬分類:
Network/ACE
這個示例將用ACE Proactor來實現(xiàn)echo server
代碼量差不多只有完成端口的1/3
ACE Reactor的實現(xiàn)
http://m.shnenglu.com/sandy/archive/2006/02/17/3308.html
完成端口的實現(xiàn)
http://m.shnenglu.com/sandy/archive/2007/06/06/25670.html當然如果你理解了完成端口，ACE Proactor就很簡單了
大概的過程是這個樣子的
1.使用ACE_Asynch_Acceptor來建立服務器端的socket,然后bind,listen等待Accept并創(chuàng)建IOCP
(相對于ACE Reactor的ACE_Acceptor）
2.當出現(xiàn)完成事件（GetQueuedCompletionStatus）返回時，調用用戶的處理接口來處理，這是一個派生于ACE_Service_Handler的子類
（相對于ACE Reactor的ACE_Svc_Handler）
ACE Reactor和Proactor最大的不同就是
ACE Reactor在處理讀寫事件的時候，要使用peer().recv/send來接收/發(fā)送數(shù)據(jù)。
而ACE Proactor在處理讀寫事件的時候,數(shù)據(jù)已經被讀到用戶的緩沖區(qū)了，典型的"先斬后奏"
ACE Proactor比ACE Reactor麻煩的地方是需要手工去投遞異步讀寫的請求。
可以通過下面的代碼來去體會。
==========代碼===============
#ifdef _DEBUG
#pragma comment(lib,"ACED.lib")
#else
#pragma comment(lib,"ACE.lib")
#endif
#include "ace/OS_main.h"
#include "ace/OS_NS_sys_socket.h"
#include "ace/ACE.h"
#include "ace/Service_Object.h"
#include "ace/Asynch_IO.h"
#include "ace/Proactor.h"
#include "ace/message_block.h"
#include "ace/Asynch_Acceptor.h"
#include "ace/INET_Addr.h"
class Echo_Service : public ACE_Service_Handler
{
public:
~Echo_Service ()
{
if (this->handle () != ACE_INVALID_HANDLE)
ACE_OS::closesocket (this->handle ());
}
virtual void open (ACE_HANDLE h, ACE_Message_Block&)
{
this->handle (h);
if (this->reader_.open (*this) != 0 || this->writer_.open (*this) != 0 )
{
ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"),ACE_TEXT ("Echo_Service open")));
delete this;
return;
}
ACE_Message_Block *mb;
ACE_NEW_NORETURN (mb, ACE_Message_Block (1024));
if (this->reader_.read (*mb, mb->space ()) != 0)
{
ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"),ACE_TEXT ("Echo_Service begin read")));
mb->release ();
delete this;
return;
}
}
virtual void handle_read_stream(const ACE_Asynch_Read_Stream::Result &result)
{
ACE_Message_Block &mb = result.message_block ();
if (!result.success () || result.bytes_transferred () == 0)
{
mb.release ();
delete this;
}
else
{
if (this->writer_.write (mb, mb.length ()) == -1)
{
ACE_ERROR ((LM_ERROR,
ACE_TEXT ("%p\n"),
ACE_TEXT ("starting write")));
mb.release ();
}
else
{
ACE_Message_Block *new_mb;
ACE_NEW_NORETURN (new_mb, ACE_Message_Block (1024));
this->reader_.read (*new_mb, new_mb->space ());
}
}
return;
}
virtual void handle_write_stream
(const ACE_Asynch_Write_Stream::Result &result)
{
ACE_Message_Block &mb = result.message_block ();
mb.release ();
return;
}
private:
ACE_Asynch_Read_Stream reader_;
ACE_Asynch_Write_Stream writer_;
};
typedef ACE_Asynch_Acceptor<Echo_Service> MyAcceptor;
int main(int argc, char *argv[])
{
ACE_INET_Addr addr(1500);
MyAcceptor server;
if(server.open(addr)==-1)
{
ACE_DEBUG ((LM_DEBUG,
ACE_TEXT ("(%P|%t) %p\n"),
ACE_TEXT ("bind failed")));
return 1;
}
while(1){
ACE_Proactor::instance ()->proactor_run_event_loop ();
}
return 0;
}