实现一个可限制最大连接数的Proactor服务器

在服务器程序实现中，一般要求能够限制服务器的最大连接数，这主要是从服务器的性能方面考虑，当过多的连接到来时，服务器虽然能够处理，但效率非常低下，也就会出现卡机的现象。

在用Proactor框架实现的服务器中可以很容易地做到这一点。ACE_Asynch_Acceptor类有一个虚方法：make_handler()，默认情况下是new一个服务处理对象出来，我们可以让他在适当的时候返回一个空值，这样，新的连接便被拒绝了。

另外为了避免每次都完全构造一个全新的服务处理对象，这里还可以使用预分配的方法，一开始便创建出所有的service对象，当有连接到来时，直接取一个未使用的分配给该连接，当连接关闭时，同时也不delete该对象，而是加到空闲队列中，等下一次有新连接到来时再使用。

实现的代码如下：

#include <ace/Os_main.h>
#include <ace/Asynch_Acceptor.h>
#include <ace/Proactor.h>
#include <list>

#define LISTEN_PORT 5222 // 服务器监听端口
#define MAX_CONNS 2 // 最大连接数

class HA_Proactive_Acceptor;

class HA_Proactive_Service : public ACE_Service_Handler
{
public:
virtual ~HA_Proactive_Service ();

void init( HA_Proactive_Acceptor * acceptor );

virtual void open (ACE_HANDLE h, ACE_Message_Block&);
virtual void handle_read_stream (const ACE_Asynch_Read_Stream::Result &result);
virtual void handle_write_stream (const ACE_Asynch_Write_Stream::Result &result);

private:
ACE_Asynch_Read_Stream reader_;
ACE_Asynch_Write_Stream writer_;

HA_Proactive_Acceptor * acceptor_;
};

class HA_Proactive_Acceptor : public ACE_Asynch_Acceptor<HA_Proactive_Service>
{
public:
HA_Proactive_Acceptor();

void free_handler( HA_Proactive_Service * service );

private:
virtual HA_Proactive_Service *make_handler (void);
void init_handlers();

private:
typedef std::list<HA_Proactive_Service *> T_Handler_List;
T_Handler_List handler_list_;
};

HA_Proactive_Service::~HA_Proactive_Service ()
{
if (this->handle () != ACE_INVALID_HANDLE)
ACE_OS::closesocket (this->handle ());
}

void HA_Proactive_Service::init( HA_Proactive_Acceptor * acceptor )
{
this->acceptor_ = acceptor;
}

void HA_Proactive_Service::open (ACE_HANDLE h, ACE_Message_Block&)
{
this->handle (h);
if (this->reader_.open (*this) != 0 || this->writer_.open (*this) != 0 )
{
ACE_OS::closesocket (this->handle ());
this->acceptor_->free_handler( this );
return;
}

ACE_Message_Block *mb;
ACE_NEW_NORETURN (mb, ACE_Message_Block (1024));
if (this->reader_.read (*mb, mb->space ()) != 0)
{
mb->release ();
ACE_OS::closesocket (this->handle ());
this->acceptor_->free_handler( this );
}
}

void HA_Proactive_Service::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 ();
ACE_OS::closesocket (this->handle ());
this->acceptor_->free_handler( this );
}
else
{
ACE_DEBUG ((LM_DEBUG, ACE_TEXT("Received Data : %c/n"), *mb.rd_ptr()));
mb.release();

ACE_Message_Block *new_mb;
ACE_NEW_NORETURN (new_mb, ACE_Message_Block (1024));
this->reader_.read (*new_mb, new_mb->space ());
}
}

void HA_Proactive_Service::handle_write_stream (const ACE_Asynch_Write_Stream::Result &result)
{
result.message_block ().release ();
}

HA_Proactive_Acceptor::HA_Proactive_Acceptor() : ACE_Asynch_Acceptor<HA_Proactive_Service>()
{
init_handlers();
}

void HA_Proactive_Acceptor::free_handler( HA_Proactive_Service * service )
{
this->handler_list_.push_back( service );
}

HA_Proactive_Service * HA_Proactive_Acceptor::make_handler (void)
{
if( this->handler_list_.empty() )
return 0;

HA_Proactive_Service * service = this->handler_list_.front();
this->handler_list_.pop_front();

return service;
}

void HA_Proactive_Acceptor::init_handlers()
{
for( int i = 0; i < MAX_CONNS; ++i )
{
HA_Proactive_Service * service;
ACE_NEW( service, HA_Proactive_Service );
service->init( this );
this->handler_list_.push_back( service );
}
}

int ACE_TMAIN (int, ACE_TCHAR *[])
{
ACE_INET_Addr listen_addr( LISTEN_PORT );
HA_Proactive_Acceptor aio_acceptor;
if (0 != aio_acceptor.open (listen_addr, 0, 0, 5, 1, 0, 0))
ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p/n"), ACE_TEXT ("acceptor open")), 1);

ACE_Proactor::instance ()->proactor_run_event_loop ();

return 0;
}