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

在服务器端使用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方法中释放资源的相应代码。