反应器组件 ACE_Reactor

6.1 反应器组件 ACE_Reactor

反应器的基本原理是:

针对关心的某个事件写一个事件处理器(event_handler). 将该事件处理器登记到反应器中(同时指明关心的事件).

然后反应器会自动检测事件的发生. 并调用预先登记的事件处理器中的回调函数.

所以. 用户要做的工作就是:

创建事件处理器.

在反应器上登记该处理器. 告诉反应器它对某个事件有兴趣.

6.2 事件处理器

在ACE中. 反应器是ACE_Reactor类的单件对象(因为程序中通常只需要一个反应器).

反应器提供了登记/撤销 事件处理器的接口. register_handler() / remove_handler() .

这些接口要求 事件处理器必须是 ACE_Event_Handler 类型的. 所以我们的事件处理器类必须从该类继承.

在ACE_Event_Handler类中定义了一些类似 "handle_***" 的回调方法. 我们必须在派生类中重写我们敢兴趣的.

在反应器需要检测某个I/O句柄上是否有事件时. 需要知道原始句柄. 这样就需要重写事件处理器类的get_handle()函数.

下边是ACE_Event_Handler 中声明的钩子函数:

handle_signal()当在反应器上登记的信号发生时. 反应器回调该函数. (不懂)

handle_input()当来自I/O设备的输入可用时. 反应器自动回调该方法.

handle_exception() 当在反应器上登记的异常事件发生时. (不懂)

handle_timeout()当在反应器上登记的定时器超时的时候. 回调该方法.

handle_output()当在IO设备上的输出可用时. 回调该方法.

6.2.1 登记事件处理器

使用 ACE_Reactor 类的 register_handler() 函数. 这个函数有好几个重载形式.

该函数有个参数用来指出感兴趣的事件. 它可以是下边一些常量(定义在ACE_Event_handler类中):

READ_MASK句柄上有数据可读时 回调 handle_input()

WRITE_MASK句柄上可写时 回调 handle_output()

TIMER_MASK回调 handle_close() 不懂怎么用...

ACCEPT_MASK有来自客户端的新的连接请求时 回调 handle_input()

CONNECT_MASK建立连接时 回调 handle_input()

DONT_CALL它用在显式拆除事件处理器的remove_handler()函数中. 表示拆除前不调用 handler_close() 函数.

6.2.2 拆除事件处理器

当不在需要处理某个事件时. 需要把对应的事件处理器从反应器中拆除.

有两种拆除事件处理器的办法:

一种是隐式的自动拆除. 当事件处理器类中的 handle_*** 方法返回的int 小于0 时. 反应器会自动调用事件处理器

的Handle_close()方法. 并把事件处理器拆除.

另一种是显式拆除. 即调用 ACE_reactor::remove_handler(). 这也会调用事件处理器的handle_close(). 然后拆除.

不过. 如果你不需要调用handle_close(). 可以给remove_handler()传递参数 ACE_Event_Handler::DONT_CALL .

具体例子在后边会给出.

6.3 通过反应器进行事件处理

6.3.1 I/O事件多路分离

////////////////////////////////////

// 一个使用反应器的例子.

// (注意这里的例子不是71页那个. 因为那个例子小弟么看懂ing).

// 来自 <<ACE程序员教程>> 75页.

// 用反应器来实现一个tcp的服务器. 连接的监听. 以及在接受的新连接的上的读取事件. 都在反应器中进行.

///////////////////////////////////

//服务器端

#include "ace/Reactor.h"

#include "ace/Event_Handler.h"

#include "ace/SOCK_Acceptor.h"

const int PORT_NO = 19999; //服务器监听的端口号.

typedef ACE_SOCK_Acceptor Acceptor;

class My_Accept_Handler;

// 一个事件处理器类.

//处理其中维护的TCP流对象 peer_ 上发生的输入事件.

class My_Input_Handler : public ACE_Event_Handler {

public:

// 重写基类中的handle_input() 函数.

int handle_input(ACE_HANDLE) { //接口中这个参数的作用是??

peer_.recv_n(data,12); //书中是 peer().recv_n(data,12); 但不知peer()从何而来....

// 输出 data...

return 0; //该函数的返回值如果小于0. 则反应器会在调用完后撤销该事件处理器的登记.

}

// 下边的函数会被反应器使用. 来得到会发生事件的底层句柄.

ACE_HANDLE get_handle() const {

return peer_.get_handle(); // 取得TCP流对象的原始句柄

}

private:

ACE_SOCK_Stream peer_;

char data[12];

};

// 另一个事件处理器类. 处理接受器收到的连接的事件

class My_Accept_Handler : public ACE_Event_Handler {

public :

My_Accept_Handler(ACE_Addr & addr) {

open(addr);

}

int open(ACE_Addr &addr) {

peer_acceptor.open(addr);// 接受器的 open() 函数 和构造函数中直接传递addr 都可以使接受器监听该地址.

return 0;

}

// 重写基类中的handle_input()

int handle_input(ACE_HANDLE handle) {

My_Input_Handler *eh = new My_Input_Handler; // 刚才定义的那个类.

if (peer_acceptor.accept( //接受一个连接

eh->peer_,//一个要捆绑的ACE_SOCK_Stream对象. (peer_是私有的. 这里本来不能直接访问. 偷懒一下马马虎虎..)

0,//这个参数用来返回前来连接的客户的地址.

0,//超时时间

1) //这个参数不懂ing..

== -1)

ACE_DEBUG((LM_ERROR,"Error in connection/n"));

//然后给反应器中增加一个事件处理器. 来处理这个新连接上的输入事件.

ACE_Reactor::instance()->register_handler(eh,ACE_Event_Handler::READ_MASK);

return -1; //返回负数.表示希望反应器在处理一个连接请求后. 自动注销本处理器.

}

// 要重写基类中的get_handle() 函数. 来帮助反应器取得内部句柄

ACE_HANDLE get_handle() const {

return peer_acceptor.get_handle();

}

private:

acceptor peer_acceptor; //接受器

};

int main() {

ACE_INET_Addr addr(PORT_NO);

My_Accept_Handler * eh = new My_Accept_Handler(addr);

// 登记eh到反应器中

ACE_Reactor::instance() -> register_handler(eh,ACE_Event_Handler::ACCEPT_MASK);

while(1)

ACE_Reactor::instance() -> handle_events(); //处理事件

}