ACE高效PROACTOR编程框架一ClientHandle

简介:

1、WIN32下面用proactor可以达到几乎RAW IOCP的效率,由于封装关系,应该是差那么一点。

客户端处理类的常规写法:
//处理客户端连接消息
class ClientHandler : public ACE_Service_Handler
{
public:
 /**构造函数
  *
  *
  */
 ClientHandler(unsigned int client_recv_buf_size=SERVER_CLIENT_RECEIVE_BUF_SIZE)
  :_read_msg_block(client_recv_buf_size),_io_count(0)
 {
 }


 ~ClientHandler(){}

 /**
  *初始化,因为可能要用到ClientHandler内存池,而这个池又不一定要用NEW
  */
 void init();

 /**清理函数,因为可能要用到内存池
  *
  */
 void fini();

//检查是否超时的函数

 void check_time_out(time_t cur_time);

public:

 /**客户端连接服务器成功后调用
  *
  * \param handle 套接字句柄
  * \param &message_block 第一次读到的数据(未用)
  */

//由Acceptor来调用!!!
 virtual void open (ACE_HANDLE handle,ACE_Message_Block &message_block);

 /**处理网络读操作结束消息
  *
  * \param &result 读操作结果
  */
 virtual void handle_read_stream (const ACE_Asynch_Read_Stream::Result &result);


 /**处理网络写操作结束消息
  *
  * \param &result 写操作结果
  */
 virtual void handle_write_stream (const ACE_Asynch_Write_Stream::Result &result);

private:

//**生成一个网络读请求
  *
  * \param void 
  * \return 0-成功,-1失败
  */
 int  initiate_read_stream  (void);

 /**生成一个写请求
  *
  * \param mb 待发送的数据
  * \param nBytes 待发送数据大小
  * \return 0-成功,-1失败
  */
 int  initiate_write_stream (ACE_Message_Block & mb, size_t nBytes );
 
 /**
  *
  * \return 检查是否可以删除,用的是一个引用计数。每一个外出IO的时候+1,每一个IO成功后-1
  */
 int check_destroy();
 
 //异步读
 ACE_Asynch_Read_Stream _rs;

 //异步写
 ACE_Asynch_Write_Stream _ws;

 //接收缓冲区只要一个就够了,因为压根就没想过要多读,我直到现在也不是很清楚为什么要多读,多读的话要考虑很多问题
 ACE_Message_Block _read_msg_block;

 //套接字句柄,这个可以不要了,因为基类就有个HANDLER在里面的。
 //ACE_HANDLE _handle;

 //一个锁,客户端反正有东东要锁的,注意,要用ACE_Recursive_Thread_Mutex而不是用ACE_Thread_Mutex,这里面是可以重入的,而且在WIN32下是直接的EnterCriticalSection,可以达到很高的效率
 ACE_Recursive_Thread_Mutex _lock;
 
 //在外IO数量,其实就是引用计数啦,没啥的。为0的时候就把这个东东关掉啦。
 long _io_count;

//检查超时用的,过段时间没东东就CLOSE他了。

 time_t _last_net_io;

private:

//本来想用另外一种模型的,只用1个或者2个外出读,后来想想,反正一般内存都是足够的,就不管了。

 //ACE_Message_Block _send_msg_blocks[2];

 //ACE_Message_Block &_sending_msg_block;

 //ACE_Message_Block &_idle_msg_block;

private:
 
public:
//TODO:move to prriva and use friend class!!!

//只是为了效率更高,不用STL的LIST是因为到现在我没有可用的Node_Allocator,所以效率上会有问题。
 ClientHandler *_next;

 ClientHandler *next(){return _next;}

 void next(ClientHandler *obj){_next=obj;}

};


//这是具体实现,有些地方比较乱,懒得管了,锁的有些地方不对。懒得改了,反正在出错或者有瓶颈的时候再做也不迟。

void ClientHandler::handle_read_stream (const ACE_Asynch_Read_Stream::Result &result)
{
 _last_net_io=ACE_OS::time(NULL);
 int byterecved=result.bytes_transferred ();
 if ( (result.success ()) && (byterecved != 0))
 {
  //ACE_DEBUG ((LM_DEBUG,  "Receiver completed:%d\n",byterecved));

//处理完数据
  if(handle_received_data()==true)
  {
   //ACE_DEBUG ((LM_DEBUG,  "go on reading...\n"));

//把东东推到头部,处理粘包
   _read_msg_block.crunch();
   initiate_read_stream();
  }
 }

//这个地方不想用ACE_Atom_op,因为反正要有一个锁,而且一般都会用锁,不管了。假如不在意的话,应该直接用ACE_Atom_Op以达到最好的效率

 {
  ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
  _io_count--;
 }
 check_destroy ();
}

void ClientHandler::init()
{

//初始化数据,并不在构造函数里做。
 _last_net_io=ACE_OS::time(NULL);
 _read_msg_block.rd_ptr(_read_msg_block.base());
 _read_msg_block.wr_ptr(_read_msg_block.base());
 this->handle(ACE_INVALID_HANDLE);
}

bool ClientHandler::handle_received_data()
{

...........自己处理
 return true;
}


//==================================================================
void ClientHandler::handle_write_stream (const ACE_Asynch_Write_Stream::Result &result)
{
 //发送成功,RELEASE掉
 //这个不可能有多个RELEASE,直接XX掉
 //result.message_block ().release ();
 MsgBlockManager::get_instance().release_msg_block(&result.message_block());

 {
  ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
  _io_count--;
 }
 check_destroy ();
}

//bool ClientHandler::destroy () 
//{
// FUNC_ENTER;
// ClientManager::get_instance().release_client_handle(this);
// FUNC_LEAVE;
// return false ;
//}


int  ClientHandler::initiate_read_stream  (void)
{
 ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);

//考虑到粘包的呀
 if (_rs.read (_read_msg_block, _read_msg_block.space()) == -1)
 {
  ACE_ERROR_RETURN ((LM_ERROR,"%p\n","ACE_Asynch_Read_Stream::read"),-1);
 }
 _io_count++;
 return 0;
}

/**生成一个写请求
*
* \param mb 待发送的数据
* \param nBytes 待发送数据大小
* \return 0-成功,-1失败
*/
int  ClientHandler::initiate_write_stream (ACE_Message_Block & mb, size_t nBytes )
{
 ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
 if (_ws.write (mb , nBytes ) == -1)
 {
  mb.release ();
  ACE_ERROR_RETURN((LM_ERROR,"%p\n","ACE_Asynch_Write_File::write"),-1);
 }
 _io_count++;
 return 0;
}

void ClientHandler::open (ACE_HANDLE handle,ACE_Message_Block &message_block)
{
 //FUNC_ENTER;
 _last_net_io=ACE_OS::time(NULL);
 _io_count=0;
 if(_ws.open(*this,this->handle())==-1)
 {
  ACE_ERROR ((LM_ERROR,"%p\n","ACE_Asynch_Write_Stream::open"));
 }
 else if (_rs.open (*this, this->handle()) == -1)
 {
  ACE_ERROR ((LM_ERROR,"%p\n","ACE_Asynch_Read_Stream::open"));
 }
 else
 {
  initiate_read_stream ();
 }

 check_destroy();
 //FUNC_LEAVE;
}

void ClientHandler::fini()
{
}

void ClientHandler::check_time_out(time_t cur_time)
{
 //ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
 //ACE_DEBUG((LM_DEBUG,"cur_time is %u,last io is %u\n",cur_time,_last_net_io));

 //检测是否已经为0了
 if(this->handle()==ACE_INVALID_HANDLE)
  return;
 if(cur_time-_last_net_io>CLIENT_TIME_OUT_SECONDS)
 {
  ACE_OS::shutdown(this->handle(),SD_BOTH);
  ACE_OS::closesocket(this->handle());
  this->handle(ACE_INVALID_HANDLE);
 }
}

int ClientHandler::check_destroy()
{
 {
  ACE_Guard<ACE_Recursive_Thread_Mutex> locker (_lock);
  if (_io_count> 0)
   return 1;
 }
 ACE_OS::shutdown(this->handle(),SD_BOTH);
 ACE_OS::closesocket(this->handle());
 this->handle(ACE_INVALID_HANDLE);

//这个地方给内存池吧。
 ClientManager::get_instance().release_client_handle(this);
 //delete this;
 return 0;
}

目录
相关文章
|
24天前
|
网络协议 安全 Java
Java网络编程实战:构建高效稳定的网络通信
【4月更文挑战第2天】Java网络编程涉及Socket编程,基于TCP(可靠,面向连接)和UDP(不可靠,无连接)协议。Socket类和ServerSocket类用于TCP,而DatagramSocket和DatagramPacket处理UDP。高效通信涉及线程管理、选择合适的IO模型(如NIO)、利用缓冲区及确保网络安全,如使用SSL/TLS。适用于Web服务器、文件传输等场景。
Java网络编程实战:构建高效稳定的网络通信
|
3月前
|
网络协议 Linux 应用服务中间件
高性能网络编程 - 解读5种I/O模型
高性能网络编程 - 解读5种I/O模型
65 0
|
3月前
Netty Review - 借助SimpleTalkRoom初体验异步网络编程的魅力
Netty Review - 借助SimpleTalkRoom初体验异步网络编程的魅力
39 0
|
6月前
|
SQL 分布式计算 算法
【大数据处理框架】Spark大数据处理框架,包括其底层原理、架构、编程模型、生态圈
【大数据处理框架】Spark大数据处理框架,包括其底层原理、架构、编程模型、生态圈
239 0
|
开发框架 负载均衡 .NET
四、分布式编程概述
四、分布式编程概述
184 0
|
数据采集 网络协议 应用服务中间件
多线程、事件驱动与推荐引擎框架选型
  事件驱动编程是一种编程范式,这里程序的执行流由外部事件来决定。它的特点是包含一个事件循环,当外部事件发生时使用回调机制来触发相应的处理。多线程是另一种常用编程范式,并且更容易理解。   高性能通用型C++网络框架 Nebula 是基于事件驱动的多进程网络框架(适用于即时通讯、数据采集、实时计算、消息推送等应用场景),已有即时通讯、埋点数据采集及实时分析的生产应用案例。
2363 0
|
Java 数据库 算法
《Akka应用模式:分布式应用程序设计实践指南》读书笔记9
性能   这也是一个比较大的问题,因为性能不一定是Akka本身的问题,还可能是你代码写的有问题。   优化的第一步就是找出性能的瓶颈,隔离出应用程序里面比较耗时的部分,然后尝试对其优化,减少需要耗费的时间成本。
1636 0
|
缓存 运维 数据库
《Akka应用模式:分布式应用程序设计实践指南》读书笔记8
可用性   简单点来说就是系统能否正常使用。如果系统能够及时响应一个请求,则认为是可用的;如果响应时间过长或者根本不响应,则是不可用的。系统在停机或超载时是不可用的。一般用系统正常运行时长的百分比来计量系统的可用性,例如常常用N个9表示系统的可用性。
1884 0
|
存储 缓存 NoSQL
《Akka应用模式:分布式应用程序设计实践指南》读书笔记6
一致性和可扩展性   一致性是系统内比较复杂的属性,它会随着系统的变化而变化。简单来说,一致性就是数据保持一致,在分布式系统中,可以理解为多个节点中数据的值是一致的。一旦系统具有并行性(分布式只是并行的一种表现),保持一致性就变得困难了,毕竟需要协调全局状态。
1367 0