贝贝花花包包店,精品555皮具,钱夹,皮夹

字体:  

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

peakzhang 发表于: 2008-9-09 22:43 来源: ACE 开发者

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

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

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

实现的代码如下:

CODE:


#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;
}