采用 ACE Reactor 实现服务程序例子

此文版权属于作者所有，任何人、媒体或者网站转载、借用都必须征得作者本人同意！

ACE 使用方法及例子，网上有不少，下面贴一段我写的采用 ACE Reactor 模式写的 echo 服务的例子代码，通过例子可以看出，采用 ACE 开发多客户端的服务程序那是相当简单的！

代码中，handle_input(…)和 handle_output(…)都会对 _bufs 进行操作，因为这两个函数都是运行在 reactor 的线程里，不会冲突，所以没有必要对 _bufs 的操作进行锁操作。

/* $Id: cpp.tpl 3412 2009-11-14 14:23:44Z luozhiyong $ */

/**

* \file ACEReactorSvrSample.cpp

*

* \brief 采用ACE Reactor 实现服务程序例子

*

* \version $Rev: 3412 $

* \author

* \date 2009 年09月08日08:17:10

*

* \note 修改历史：<br>

* <table>

* <tr><th> 日期</th><th>修改人</th><th>内容</th></tr>

* <tr><td>2009-9-8</td><td> </td><td>创建初稿</td>

* </tr>

* </table>

*/

#include <ace/Message_Block.h>

#include <ace/Svc_Handler.h>

#include <ace/SOCK_Acceptor.h>

#include <ace/Acceptor.h>

#include <ace/Select_Reactor.h>

#include <list>

#include <string>

#ifdef _DEBUG

# define ACE_RT_OPT "d"

#else

# define ACE_RT_OPT

#endif

#if defined _DLL

# define ACE_LIB_THREAD_OPT

#else

# define ACE_LIB_THREAD_OPT "s"

#endif

#pragma comment( lib,"ACE" ACE_LIB_THREAD_OPT ACE_RT_OPT ".lib")

class EchoService

: public ACE_Event_Handler

{

public:

typedef ACE_SOCK_STREAM stream_type;

typedef EchoService my_type;

typedef ACE_Acceptor< my_type, ACE_SOCK_ACCEPTOR> acceptor_type;

EchoService()

{

printf( "EchoService 创建\n");

}

~ EchoService()

{

printf( "EchoService 销毁\n");

}

// 响应socket 已经打开，连接已经建立事件

int open( void*)

{

// 注册读事件

if ( reactor()-> register_handler( this,ACE_Event_Handler:: READ_MASK))

{

// 无法注册handler

return -1;

}

// 注册写事件

if ( reactor()-> register_handler( this,ACE_Event_Handler:: WRITE_MASK))

{

// 无法注册handler

return -1;

}

// 取消写事件，等待有数据时唤醒

reactor()-> cancel_wakeup( this,ACE_Event_Handler:: WRITE_MASK);

printf( "EchoService 已打开\n");

return 0;

}

// 响应有数据可读事件

int handle_input( ACE_HANDLE)

{

char buf[24];

ssize_t c = _peer. recv( buf,sizeof( buf) - 1);

if ( c == 0)

{

// 连接已经关闭

return -1;

}

_bufs. push_back( std:: string( buf,c));

if ( _bufs. size() == 1)

{

// 缓冲区尺寸为1 说明原来缓冲区为空，写事件是取消的，这里唤醒它

reactor()-> schedule_wakeup( this,ACE_Event_Handler:: WRITE_MASK);

}

return 0;

}

// 响应可以发送数据了事件

int handle_output( ACE_HANDLE)

{

while (! _bufs. empty())

{

std:: string& buf(* _bufs. begin());

char const* s( buf. c_str());

char const* const e( s + buf. size());

while ( s != e)

{

ssize_t c( _peer. send( s,e - s));

if ( c == -1 || c == 0)

{

// 发送不成功不论发送过程中是否发生阻塞，

if ( ACE_OS:: last_error() == EWOULDBLOCK)

{

// 输出缓冲区满，无法再发送数据了（如果你还是继续发送数据，发送会阻塞的）

break;

} else{

// 连接已关闭

break;

}

} else{

s += c;

}

}

if ( s == e)

{

_bufs. pop_front();

} else{

buf = std:: string( s,e - s);

break;

}

}

if ( _bufs. empty())

{

// 缓冲区为空，取消写事件监听

reactor()-> cancel_wakeup( this,ACE_Event_Handler:: WRITE_MASK);

}

// 不论发送是否成功都返回0，因为，如果发送失败，handle_input 也会发生读失败事件，

// 错误处理有handle_input 返回-1 来触发

return 0;

}

int handle_close( ACE_HANDLE = ACE_INVALID_HANDLE,ACE_Reactor_Mask mask = ACE_Event_Handler:: ALL_EVENTS_MASK)

{

if ( mask == ACE_Event_Handler:: WRITE_MASK)

return 0;

_peer. close();

delete this;

return 0;

}

// 这个函数主要给reactor::register_handler 时使用的

ACE_HANDLE get_handle () const

{

return _peer. get_handle();

}

// 这个函数主要给acceptor 使用的

stream_type& peer()

{

return _peer;

}

// 这个函数主要给acceptor 使用的

int close ( u_long = 0)

{

return handle_close();

}

private:

stream_type _peer;

std:: list< std:: string> _bufs;

};

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

{

u_short port = 20001;

ACE_Reactor:: instance( new ACE_Reactor( new ACE_Select_Reactor,true));

EchoService:: acceptor_type acceptor;

ACE_INET_Addr svrAddr( port);

if ( acceptor. open( svrAddr))

{

fprintf( stderr," 服务打开失败：%s\n",ACE_OS:: strerror( ACE_OS:: last_error()));

return 1;

} else{

fprintf( stdout," 服务已打开，端口为：%u\n",port);

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

return 0;

}

}