ACE反应器(Reactor)模式(3)

本贴转自http://www.cnblogs.com/TianFang/category/78013.html 作者:天方

在服务器端使用Reactor框架


使用Reactor框架的服务器端结构如下：

服务器端注册两种事件处理器，ClientAcceptor和ClientService ，ClientService类负责和客户端的通信，每一个ClientService对象对应一个客户端的Socket连接。
ClientAcceptor专门负责被动接受客户端的连接，并创建ClientService对象。这样，在一个N个Socket连接的服务器程序中，将存在1个ClientAcceptor对象和N个ClientService对象。

整个服务器端流程如下：

首先创建一个ClientAcceptor对象，该对象在Reactor上注册ACCEPT_MASK事件，Reactor将自动在监听端口建立Socket监听。

如果有对该端口的Socket连接时，Reactor将自动回调handle_input方法，ClientAcceptor重载此方法，并创建一个ClientService对象，用于处理和Client的通信。

ClientService对象根据服务器的具体功能实现，其处理过程和客户端程序类似，注册相应的回调事件并分发即可。

代码如下：

#include
<ace/OS.h>

#include
<ace/Reactor.h>

#include
<ace/SOCK_Connector.h>

#include
<ace/SOCK_Acceptor.h>

#include
<ace/Auto_Ptr.h>

class
ClientService : public
ACE_Event_Handler

{

public
:

ACE_SOCK_Stream &peer (void
) { return
this
->sock_; }

int
open (void
)

{

//注册读就绪回调函数

return
this
->reactor ()->register_handler(this
, ACE_Event_Handler::READ_MASK);

}

virtual
ACE_HANDLE get_handle (void
) const
{ return
this
->sock_.get_handle (); }

virtual
int
handle_input (ACE_HANDLE fd )

{

//一个简单的EchoServer，将客户端的信息返回

int
rev = peer().recv(buf,100);

if
(rev<=0)

return
-1;

peer().send(buf,rev);

return
0;

}

// 释放相应资源

virtual
int
handle_close (ACE_HANDLE, ACE_Reactor_Mask mask)

{

if
(mask == ACE_Event_Handler::WRITE_MASK)

return
0;

mask = ACE_Event_Handler::ALL_EVENTS_MASK |

ACE_Event_Handler::DONT_CALL;

this
->reactor ()->remove_handler (this
, mask);

this
->sock_.close ();

delete
this
; //socket出错时，将自动删除该客户端，释放相应资源

return
0;

}

protected
:

char
buf[100];

ACE_SOCK_Stream sock_;

};

class
ClientAcceptor : public
ACE_Event_Handler

{

public
:

virtual
~ClientAcceptor (){this
->handle_close (ACE_INVALID_HANDLE, 0);}

int
open (const
ACE_INET_Addr &listen_addr)

{

if
(this
->acceptor_.open (listen_addr, 1) == -1)

{

ACE_OS::printf("open port fail"
);

return
-1;

}

//注册接受连接回调事件

return
this
->reactor ()->register_handler(this
, ACE_Event_Handler::ACCEPT_MASK);

}

virtual
ACE_HANDLE get_handle (void
) const

{ return
this
->acceptor_.get_handle (); }

virtual
int
handle_input (ACE_HANDLE fd )

{

ClientService *client = new
ClientService();

auto_ptr<ClientService> p (client);

if
(this
->acceptor_.accept (client->peer ()) == -1)

{

ACE_OS::printf("accept client fail"
);

return
-1;

}

p.release ();

client->reactor (this
->reactor ());

if
(client->open () == -1)

client->handle_close (ACE_INVALID_HANDLE, 0);

return
0;

}

virtual
int
handle_close (ACE_HANDLE handle,

ACE_Reactor_Mask close_mask)

{

if
(this
->acceptor_.get_handle () != ACE_INVALID_HANDLE)

{

ACE_Reactor_Mask m = ACE_Event_Handler::ACCEPT_MASK |

ACE_Event_Handler::DONT_CALL;

this
->reactor ()->remove_handler (this
, m);

this
->acceptor_.close ();

}

return
0;

}

protected
:

ACE_SOCK_Acceptor acceptor_;

};

int
main(int
argc, char
*argv[])

{

ACE_INET_Addr addr(3000,"192.168.1.142"
);

ClientAcceptor server;

server.reactor(ACE_Reactor::instance());

server.open(addr);

while
(true
)

{

ACE_Reactor::instance()->handle_events();

}

return
0;

}

代码功能比较简单，需要注意以下几点：

这里注册事件的方式和前面的文章中方式不一样，是通过ACE_Event_Handler类的reactor()
方法设置和获取reactor的指针，比较直观和方便。前面的文章是通过ACE_Reactor::instance()来获取的一个单体reactor的指针。

当客户端Socket连接关闭时，需要释放相应资源
，需要注意一下ClientService对象的handle_close方法中释放资源的相应代码。