ACE学习（十）前摄器模式

前摄器模式

1.前摄器的原理

前摄器模式是充分使用了操作系统的支持异步I/O的特性，使用异步来代替多线程并发，因为多线程并发会带来很多性能的开销（创建、切换、同步等），而异步像更加底层的操作系统操作，性能是相对较高（为什么？待补充）

2.前摄器的优点

“与传统的反应式或同步式I/O模型不同，前摄式模型允许每个应用线程同时发起多个操作。这一设计允许单线程化应用并发地执行多个I/O操作，并且不会带来与传统的多线程化机制相关联的开销或设计复杂性”

“使用前摄器模式的主要优点是可以启动多个并发操作，并可并行运行，而不要求应用必须拥有多个线程。操作被应用异步地启动，它们在OS的I/O子系统中运行直到完成。发起操作的线程现在可以服务另外的请求了。

Completion Dispatcher可以是单线程的。当HTTP请求到达时，单个Completion Dispatcher线程解析请求，读取文件，并发送响应给客户。因为响应是被异步发送的，多个响应就有可能同时被发送。而且，同步的文件读取可以被异步的文件读取取代，以进一步增加并发的潜力。如果文件读取是被异步完成的，HTTP Handler所执行的唯一的同步操作就只剩下了HTTP协议请求解析。

前摄式模型的主要缺点是编程逻辑至少和反应式模型一样复杂。而且，前摄器模式可能会难以调试，因为异步操作常常有着不可预测和不可重复的执行序列，这就使分析和调试复杂化了。”

3.类介绍：

ACE_Service_Handler

/**

* @class ACE_Service_Handler

*

* @brief This base class defines the interface for the

* ACE_Asynch_Acceptor to call into when new connection are

* accepted.

*

* Subclasses of this class will fill in appropriate methods to

* define application specific behavior.

*/

这个是创建服务器 处理器的必须的目标类。此类中的open()挂钩方法会在新连接建立时被调用。

ACE_Asynch_Read_Stream

/**

* @class ACE_Asynch_Read_Stream

*

* @brief This class is a factory for starting off asynchronous reads

* on a stream. This class forwards all methods to its

* implementation class.

*

* Once {open} is called, multiple asynchronous {read}s can

* started using this class. An ACE_Asynch_Read_Stream::Result

* will be passed back to the {handler} when the asynchronous

* reads completes through the {ACE_Handler::handle_read_stream}

* callback.

*/

这个为在已链接的socket上发起异步读的工厂类。

Result

/**

* @class Result

*

* @brief This is the class which will be passed back to the

* {handler} when the asynchronous read completes. This class

* forwards all the methods to the implementation classes.

*

* This class has all the information necessary for the

* {handler} to uniquiely identify the completion of the

* asynchronous read.

*/

每个I/O工厂类都把Result定义为嵌在自己内部的类，用以保存改工厂发起的每个操作的结果。

4.代码分析，时间服务器回显的服务器代码如下：

#include "ace/Message_Queue.h"
#include "ace/Asynch_IO.h"
#include "ace/OS.h"
#include "ace/Proactor.h"
#include "ace/Asynch_Acceptor.h"

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

virtual void open (ACE_HANDLE h, ACE_Message_Block&)
{
// handle方法是从ACE_Service_Handler的父类ACE_Handler继承而得
// 这个方法为hook方法，在新建立连接的时候被调用
// 这个方法会保存新的socket句柄
this->handle (h);

// 在保存完socket以后，必须用open初始化I/O工厂对象reader_，为发起I/O做准备
/**这个是open方法的参数与注释
* Initializes the factory with information which will be used with
* each asynchronous call.
*
* @arg handler The ACE_Handler that will be called to handle completions
*              for operations initiated using this factory.
* @arg handle  The handle that future read operations will use.
*              If handle == @c ACE_INVALID_HANDLE,
*              ACE_Handler::handle() will be called on @ handler
*              to get the correct handle.
*
* @retval 0    for success.
* @retval -1   for failure; consult @c errno for further information.
*/

//   int open (ACE_Handler &handler,
//             ACE_HANDLE handle = ACE_INVALID_HANDLE,
//             const void *completion_key = 0,
//             ACE_Proactor *proactor = 0);
// 其中，第一个参数表示工厂操作的完成处理器，当完成I/O操作的时候Proactor框架会
// 回调这个对象，这也是处理器叫做完成处理器的原因
// ACE_Service_Handler是ACE_Handler的后代，即是读操作也是写操作的完成处理器所以
// *this被用作处理器的参数
// 没有句柄传入，I/O工厂会调用ACE_Service_Handler::handler()方法来取得ACE_HANDLE
// ACE_Proactor也是默认的，使用的为一个进程范围内的一个ACE_Proactor对象

// 开始这里写错成&&，而&&第一部分是真的话不会进行下一部分判断也就是写工厂没有初始化
if ((this->reader_.open (*this) != 0) || (this->writer_.open(*this) != 0))
{
ACE_ERROR ((LM_ERROR, ACE_TEXT ("%p\n"),
ACE_TEXT ("HA_Proactive_Service open")));
delete this;
}

// open方法的最后一件事情就是调用I/O对象的read()方法来在socket上发起一个异步读操作
ACE_Message_Block *mb;
ACE_NEW_NORETURN(mb, ACE_Message_Block(1024));

if (this->reader_.read (*mb, mb->space ()) != 0)
{
ACE_OS::printf("Begin read fail\n");
delete this;
}
}

// 异步读完成后会调用此函数，此函数为ACE_Handler中的接口
// result为为I/O工厂的类对象，用于保存读取的结果
virtual void 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 ();
delete this;
}
else
{
// 为字符串添加结束标记'\0'并打印出来
ACE_OS::printf("rev:\t%s\n",mb.rd_ptr());

// 给客户端写回时间
time_t now = ACE_OS::gettimeofday().sec();
char *time = ctime(&now);        //获取当前时间的字符串格式
mb.copy(time);

ACE_OS::printf(mb.rd_ptr());
if (this->writer_.write(mb, mb.length()) !=0)
{
ACE_OS::printf("Begin write fail\n");
mb.release();
}
else
{
// 发起一个新的读操作（上面open方法的最后一件事情）
ACE_Message_Block *nmb;
ACE_NEW_NORETURN(nmb, ACE_Message_Block(1024));
this->reader_.read (*nmb, nmb->space ());

}
}
}

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

private:
ACE_Asynch_Read_Stream reader_;
ACE_Asynch_Write_Stream writer_;
};

int ACE_TMAIN(int, ACE_TCHAR *[])
{
int port=3000;
ACE_Asynch_Acceptor<HA_Proactive_Service> acceptor;

/**
* @c open starts one or more asynchronous accept requests on a
* @a address. Each accept operation may optionally read an
* initial buffer from the new connection when accepted.
*/
// 初始化接收器对象，并开始接受连接
// 不是上面提到的任何一个方法
if (acceptor.open (ACE_INET_Addr (port)) == -1)
return -1;

//     while(true)
//         ACE_Proactor::instance ()->handle_events ();
// 或者
ACE_Proactor::instance()->proactor_run_event_loop();

return 0;
}

客户端代码如下：

// client.cpp
#include <iostream>
using namespace std;

#include "ace/INET_Addr.h"
#include "ace/SOCK_Stream.h"
#include "ace/SOCK_Connector.h"
#include "ace/Log_Msg.h"
#include "common/MD5.h"

int ACE_TMAIN(int, ACE_TCHAR *[])
{
ACE_INET_Addr srvr(3000, ACE_LOCALHOST);
ACE_SOCK_Connector connector;
ACE_SOCK_Stream peer;

if(-1 == connector.connect(peer, srvr))
{
ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("%p\n"), ACE_TEXT("connect")), 1);
}

int bc;
char buf[64];

for (int i=0; i<10; ++i)
{
peer.send_n("uptime\n", 7);
bc = peer.recv(buf, sizeof(buf));
write(1, buf, bc);
ACE_OS::sleep(1);
}

peer.close();
return 0;
}

http://docs.huihoo.com/ace_tao/ACE-2002-12/Part-One/Chapter-8.htm

/article/4658706.html

/article/10266514.html