1．ACE反應器框架簡介
反應器（Reactor）：用于事件多路分離和分派的體系結構模式
通常的，對一個文件描述符指定的文件或設備, 有兩種工作方式:?阻塞與非阻塞。所謂阻塞方式的意思是指, 當試圖對該文件描述符進行讀寫時,如果當時沒有東西可讀,或者暫時不可寫, 程序就進入等待狀態, 直到有東西可讀或者可寫為止。而對于非阻塞狀態,如果沒有東西可讀, 或者不可寫, 讀寫函數馬上返回,而不會等待。
在前面的章節中提到的Tcp通信的例子中，就是采用的阻塞式的工作方式：當接收tcp數據時，如果遠端沒有數據可以讀，則會一直阻塞到讀到需要的數據為止。這種方式的傳輸和傳統的被動方法的調用類似，非常直觀，并且簡單有效，但是同樣也存在一個效率問題，如果你是開發一個面對著數千個連接的服務器程序，對每一個客戶端都采用阻塞的方式通信，如果存在某個非常耗時的讀寫操作時，其它的客戶端通信將無法響應，效率非常低下。
一種常用做法是：每建立一個Socket連接時，同時創建一個新線程對該Socket進行單獨通信（采用阻塞的方式通信）。這種方式具有很高的響應速度，并且控制起來也很簡單，在連接數較少的時候非常有效，但是如果對每一個連接都產生一個線程的無疑是對系統資源的一種浪費，如果連接數較多將會出現資源不足的情況。
另一種較高效的做法是：服務器端保存一個Socket連接列表，然后對這個列表進行輪詢，如果發現某個Socket端口上有數據可讀時（讀就緒），則調用該socket連接的相應讀操作；如果發現某個Socket端口上有數據可寫時（寫就緒），則調用該socket連接的相應寫操作；如果某個端口的Socket連接已經中斷，則調用相應的析構方法關閉該端口。這樣能充分利用服務器資源，效率得到了很大提高。
在Socket編程中就可以通過select等相關API實現這一方式。但直接用這些API控制起來比較麻煩，并且也難以控制和移植，在ACE中可以通過Reactor模式簡化這一開發過程。
反應器本質上提供一組更高級的編程抽象，簡化了事件驅動的分布式應用的設計和實現。除此而外，反應器還將若干不同種類的事件的多路分離集成到易于使用的API中。特別地，反應器對基于定時器的事件、信號事件、基于I/O端口監控的事件和用戶定義的通知進行統一地處理。
ACE中的反應器與若干內部和外部組件協同工作。其基本概念是反應器框架檢測事件的發生（通過在OS事件多路分離接口上進行偵聽），并發出對預登記事件處理器（eventhandler）對象中的方法的"回調"（callback）。該方法由應用開發者實現，其中含有應用處理此事件的特定代碼。
使用ACE的反應器，只需如下幾步：
創建事件處理器，以處理他所感興趣的某事件。 在反應器上登記，通知說他有興趣處理某事件，同時傳遞他想要用以處理此事件的事件處理器的指針給反應器。 隨后反應器框架將自動地：
在內部維護一些表，將不同的事件類型與事件處理器對象關聯起來。 在用戶已登記的某個事件發生時，反應器發出對處理器中相應方法的回調。 反應器模式在ACE中被實現為ACE_Reactor類，它提供反應器框架的功能接口。
如上面所提到的，反應器將事件處理器對象作為服務提供者使用。反應器內部記錄某個事件處理器的特定事件的相關回調方法。當這些事件發生時，反應器會創建這種事件和相應的事件處理器的關聯。
事件處理器
事件處理器就是需要通過輪詢發生事件改變的對象列表中的對象，如在上面的例子中就是連接的客戶端，每個客戶端都可以看成一個事件處理器。 回調事件
就是反應器支持的事件，如Socket讀就緒，寫就緒。拿上面的例子來說，如果某個客戶端（事件處理器）在反應器中注冊了讀就緒事件，當客戶端給服務器發送一條消息的時候，就會觸發這個客戶端的數據可讀的回調函數。 在反應器框架中，所有應用特有的事件處理器都必須由ACE_Event_Handler的抽象接口類派生。可以通過重載相應的"handle_"方法實現相關的回調方法。
使用ACE_Reactor基本上有三個步驟：
創建ACE_Event_Handler的子類，并在其中實現適當的"handle_"方法，以處理你想要此事件處理器為之服務的事件類型。 通過調用反應器對象的register_handler()，將你的事件處理器登記到反應器。 在事件發生時，反應器將自動回調相應的事件處理器對象的適當的handle_"方法。 下面我就以一個Socket客戶端的例子為例簡單的說明反應器的基本用法。
?
Cpp代碼??
#include?<ace/OS.h>?? #include?<ace/Reactor.h>?? #include?<ace/SOCK_Connector.h>??? ?? #include?<string>?? #include?<iostream>?? using?namespace?std;?? ?? class?MyClient:public?ACE_Event_Handler??? {?? public:?? ????bool?open()?? ????{?? ????????ACE_SOCK_Connector?connector;?? ????????ACE_INET_Addr?addr(3000,"127.0.0.1");?? ????????ACE_Time_Value?timeout(5,0);?? ????????if(connector.connect(peer,addr,&timeout)?!=?0)?? ????????{?? ????????????cout<<endl<<"connecetd?fail";?? ????????????return?false;?? ????????}?? ????????ACE_Reactor::instance()->register_handler(this,ACE_Event_Handler::READ_MASK);?? ????????cout<<endl<<"connecetd?";?? ????????return?true;?? ????}?? ?? ????ACE_HANDLE?get_handle(void)?const?? ????{?? ????????return?peer.get_handle();?? ????}?? ?? ????int?handle_input?(ACE_HANDLE?fd)?? ????{?? ????????int?rev=0;?? ????????ACE_Time_Value?timeout(5,0);?? ????????if((rev=peer.recv(buffer,1000,&timeout))>0)?? ????????{?? ????????????buffer[rev]='\0';?? ????????????cout<<endl<<"rev:\t"<<buffer<<endl;?? ????????}?? ????????return?3;?? ????}?? ?? private:?? ????ACE_SOCK_Stream?peer;?? ????char?buffer[1024];?? };?? ?? int?main(int?argc,?char?*argv[])??? {?? ????MyClient?client;?? ????client.open();?? ?? ????while(true)?? ????{?? ????????ACE_Reactor::instance()->handle_events();??? ????}?? ?? ????return?0;??? }?? ? ?
在這個例子中，客戶端連接上服務器后，通過ACE_Reactor::instance()->register_handler(this,ACE_Event_Handler::READ_MASK)注冊了一個讀就緒的回調函數，當服務器端給客戶端發消息的時候，會自動觸發handle_input()函數，將接收到的信息打印出來。
這個例子只是為了演示反應器的基本用法，并不完善，我將在下一節中對如何在Socket通信中使用反應器做進一步的介紹。
?
在Socket編程中，常見的事件就是"讀就緒"，"寫就緒"，通過對這兩個事件的捕獲分發，可以實現Socket中的異步操作。
Socket編程中的事件處理器
在前面我們已經介紹過，在ACE反應器框架中，任何都必須派生自ACE_Event_Handler類，并通過重載其相應會調事件處理函數來實現相應的回調處理的。在Socket編程中，我們通常需要重載的函數有
handle_input()
當I/O句柄（比如UNIX中的文件描述符）上的輸入可用時，反應器自動回調該方法。 handle_output()
當I/O設備的輸出隊列有可用空間時，反應器自動回調該方法。 handle_close()
當事件處理器中的事件從Reactor中移除的時候調用。 此外，為了使Reactor能通過I/O句柄找到對應的事件處理器，還必須重載其get_handle()方法以使得Reactor建立起I/O句柄和事件處理器的關聯。
使用Reactor框架。
下面我們將以一個客戶端的程序為例，介紹如何在Socket編程中使用Reactor框架。
一．建立一個客戶端對象（事件處理器）。
客戶端對象就是一個事件處理器，其聲明如下：
?
Cpp代碼??
class?Client:public?ACE_Event_Handler?? {?? public:?? ????ACE_HANDLE?get_handle(void)?const;?? ????int?handle_input?(ACE_HANDLE?fd);?? ????int?handle_close?(ACE_HANDLE?handle,?? ACE_Reactor_Mask?close_mask);?? ????ACE_SOCK_Stream&?Peer();?? private:?? ????ACE_SOCK_Stream?peer;?? };?? ? ?
在Client端中我只關心"讀就緒"事件，故只重載了handle_input函數（大多數應用下只需要重載handle_input函數）。另外，在客戶端還保存了一個ACE_SOCK_Stream的peer對象用來進行Socket通信，同時封裝了一個Peer()函數返回它的引用。
二．重載相應回調處理函數
?
Cpp代碼??
ACE_SOCK_Stream&?Client::Peer()?? {?? ????return?peer;?? }?? ?? ACE_HANDLE?Client::get_handle(void)?const?? {?? ????return?peer.get_handle();?? }?? ?? int?Client::handle_input?(ACE_HANDLE?fd)?? {?? ????int?rev=0;?? ????if((rev?=?peer.recv(buffer,1000))>0)?? ????{?? ????????buffer[rev]='\0';?? ????????cout<<endl<<"rev:\t"<<buffer<<endl;?? ????????return?0;?? ????}?? ????else????//Socket連接發生錯誤，返回-1，在Reactor中注銷事件，觸發handle_close函數?? ????{?? ????????return?-1;?? ????}?? }?? ?? int?Client::handle_close?(ACE_HANDLE?handle,?? ?????????????????????????ACE_Reactor_Mask?close_mask)?? {?? ????cout<<endl<<"connecetd?closed";?? ????return?ACE_Event_Handler::handle_close(handle,close_mask);?? }?? ? ?
幾個函數的功能都非常簡單，這里就不多做介紹了。
三．在Reactor中注冊事件
首先讓我們來看看相應的main函數的代碼：
?
Cpp代碼??
int?main(int?argc,?char?*argv[])??? {?? ????Client?client;?? ????ACE_SOCK_Connector?connector;?? ????ACE_INET_Addr?addr(3000,"127.0.0.1");?? ????ACE_Time_Value?timeout(5,0);?? ????if(connector.connect(client.Peer(),addr,&timeout)?!=?0)?? ????{?? ????????cout<<endl<<"connecetd?fail";?? ????????return?0;?? ????}?? ?? ????ACE_Reactor::instance()->register_handler(&client,ACE_Event_Handler::READ_MASK);?? ?? while(true)?? {?? ACE_Reactor::instance()->handle_events();??? }?? ?? return?0;??? }?? ? ?
在這里可以看到，使用Reactor框架后，依然首先通過ACE_SOCK_Connector的connect函數來建立連接。建立連接后，可以通過ACE_Reactor::instance()->register_handler函數來實現Reactor的注冊，實現I/O事件和Client對象的handle_input方法相關聯，它的第一個參數是事件處理器的地址，第二個參數是事件類型，由于這里只關心讀就緒事件，故注冊的事件類型是ACE_Event_Handler::READ_MASK。
四．啟動Reactor事件循環
通過如上設置后，我們就可以通過ACE_Reactor::instance()->handle_events()啟動Reactor循環了，這樣，每當服務器端有數據發送給客戶端時，當客戶端的數據就緒時，就回觸發Client對象的handle_input函數，將接收的數據打印出來。
通常的做法是，將Reactor事件循環作為一個單獨的線程來處理，這樣就不會阻塞main函數。
五．注銷Reactor事件
Reactor事件的注銷一般有兩種方式，顯式和隱式，下面將分別給予介紹。
隱式注銷。
當Reactor捕獲事件后，會觸發相應的"handle_"處理函數，當"handle_"處理函數返回值大于或等于0時，表示處理事件成功，當返回值小于0時，表示處理事件失敗，這時Reactor會自動注銷該句柄所注冊的所有事件，并觸發handle_close函數，以執行相應的資源清理工作。
在本例中，當handle_input函數里的recv函數接收到0個數時，說明socket發生錯誤（大多為Socket連接中斷），此時返回－1，則系統自動注銷相應事件。 顯示注銷。
調用Reactor對象的remove_handler方法移除，它有兩個參數，第一個是所注冊的事件反應器對象，第二個是所要注銷的事件。 在這個示例程序里，連接方只有一個Socket連接，Reactor的優勢并沒有體現出來，但在一些網絡管理系統里，連接方需要對多個需要管理的設備（服務器端）進行連接，在這種情況下使用Reactor模式，只需要多開一個Reactor事件循環線程就能實現事件多路分發復用，并且不會阻塞，通過面向對象的回調方式管理，使用起來非常方便。
Reactor框架的另外一個常用的地方就是服務器端，一般是一個服務器端對應多個客戶端，這樣用Reactor模式能大幅提高并發能力，這方面的編程方法將在下一章給與介紹。
?
在服務器端使用Reactor框架
使用Reactor框架的服務器端結構如下：
服務器端注冊兩種事件處理器，ClientAcceptor和ClientService，ClientService類負責和客戶端的通信，每一個ClientService對象對應一個客戶端的Socket連接。ClientAcceptor專門負責被動接受客戶端的連接，并創建ClientService對象。這樣，在一個N個Socket連接的服務器程序中，將存在1個ClientAcceptor對象和N個ClientService對象。
整個服務器端流程如下：
首先創建一個ClientAcceptor對象，該對象在Reactor上注冊ACCEPT_MASK事件，Reactor將自動在監聽端口建立Socket監聽。 如果有對該端口的Socket連接時，Reactor將自動回調handle_input方法，ClientAcceptor重載此方法，并創建一個ClientService對象，用于處理和Client的通信。 ClientService對象根據服務器的具體功能實現，其處理過程和客戶端程序類似，注冊相應的回調事件并分發即可。 代碼如下：
?
Cpp代碼??
#include?<ace/OS.h>?? #include?<ace/Reactor.h>?? #include?<ace/SOCK_Connector.h>??? #include?<ace/SOCK_Acceptor.h>??? #include?<ace/Auto_Ptr.h>?? ?? class?ClientService?:?public?ACE_Event_Handler?? {?? public:?? ????ACE_SOCK_Stream?&peer?(void)?{?return?this->sock_;?}?? ?? ????int?open?(void)?? ????{?? ????????//注冊讀就緒回調函數?? ????????return?this->reactor?()->register_handler(this,?ACE_Event_Handler::READ_MASK);?? ????}?? ?? ????virtual?ACE_HANDLE?get_handle?(void)?const?{?return?this->sock_.get_handle?();?}?? ?? ????virtual?int?handle_input?(ACE_HANDLE?fd?)?? ????{?? ????????//一個簡單的EchoServer，將客戶端的信息返回?? ????????int?rev?=?peer().recv(buf,100);?? ????????if(rev<=0)?? ????????????return?-1;?? ?? ????????peer().send(buf,rev);?? ????????return?0;?? ????}?? ?? ????//?釋放相應資源?? virtual?int?handle_close?(ACE_HANDLE,?ACE_Reactor_Mask?mask)?? ????{?? ????????if?(mask?==?ACE_Event_Handler::WRITE_MASK)?? ????????????return?0;?? ????????mask?=?ACE_Event_Handler::ALL_EVENTS_MASK?|?? ????????????ACE_Event_Handler::DONT_CALL;?? ????????this->reactor?()->remove_handler?(this,?mask);?? ????????this->sock_.close?();?? ????????delete?this;????//socket出錯時，將自動刪除該客戶端，釋放相應資源?? ????????return?0;?? ????}?? ?? protected:?? ????char?buf[100];?? ????ACE_SOCK_Stream?sock_;?? };?? ?? class?ClientAcceptor?:?public?ACE_Event_Handler?? {?? public:?? ????virtual?~ClientAcceptor?(){this->handle_close?(ACE_INVALID_HANDLE,?0);}?? ?? ????int?open?(const?ACE_INET_Addr?&listen_addr)?? ????{?? ????????if?(this->acceptor_.open?(listen_addr,?1)?==?-1)?? ????????{?? ????????????ACE_OS::printf("open?port?fail");?? ????????????return?-1;?? ????????}?? ????????//注冊接受連接回調事件?? ????????return?this->reactor?()->register_handler(this,?ACE_Event_Handler::ACCEPT_MASK);?? ????}?? ?? ????virtual?ACE_HANDLE?get_handle?(void)?const?? ????{?return?this->acceptor_.get_handle?();?}?? ?? ????virtual?int?handle_input?(ACE_HANDLE?fd?)?? ????{?? ????????ClientService?*client?=?new?ClientService();?? ????????auto_ptr<ClientService>?p?(client);?? ?? ????????if?(this->acceptor_.accept?(client->peer?())?==?-1)?? ????????{?? ????????????ACE_OS::printf("accept?client?fail");?? ????????????return?-1;?? ????????}?? ????????p.release?();?? ????????client->reactor?(this->reactor?());?? ????????if?(client->open?()?==?-1)?? ????????????client->handle_close?(ACE_INVALID_HANDLE,?0);?? ????????return?0;?? ????}?? ?????? ????virtual?int?handle_close?(ACE_HANDLE?handle,?? ????????ACE_Reactor_Mask?close_mask)?? ????{?? ????????if?(this->acceptor_.get_handle?()?!=?ACE_INVALID_HANDLE)?? ????????{?? ????????????ACE_Reactor_Mask?m?=?ACE_Event_Handler::ACCEPT_MASK?|?? ????????????????ACE_Event_Handler::DONT_CALL;?? ????????????this->reactor?()->remove_handler?(this,?m);?? ????????????this->acceptor_.close?();?? ????????}?? ????????return?0;?? ????}?? ?? protected:?? ????ACE_SOCK_Acceptor?acceptor_;?? };?? ?? int?main(int?argc,?char?*argv[])??? {?? ????ACE_INET_Addr?addr(3000,"192.168.1.142");?? ????ClientAcceptor?server;?? ????server.reactor(ACE_Reactor::instance());?? ????server.open(addr);?? ?? ????while(true)?? ????{?? ????????ACE_Reactor::instance()->handle_events();??? ????}?? ?? ????return?0;??? }?? ? ?
代碼功能比較簡單，需要注意以下幾點：
這里注冊事件的方式和前面的文章中方式不一樣，是通過ACE_Event_Handler類的reactor()方法設置和獲取reactor的指針，比較直觀和方便。前面的文章是通過ACE_Reactor::instance()來獲取的一個單體reactor的指針。 當客戶端Socket連接關閉時，需要釋放相應資源，需要注意一下ClientService對象的handle_close方法中釋放資源的相應代碼。
定時器的實現
通過Reactor機制，還可以很容易的實現定時器的功能，使用方式如下。
編寫一個事件反應器，重載handle_timeout()方法，該方法是定時器的觸發時間到時，會自動觸發該方法。 通過Reactor的schedule_timer()方法注冊定時器。 啟動reacotr的handle_events()事件分發循環。 當不想使用定時器時，可以通過Reactor的cancel_timer()方法注銷定時器。 下面的代碼簡單的實現了一個定時器，并具有基本的開啟，關閉功能。
?
Cpp代碼??
#include?<ace/OS.h>?? #include?<ace/Reactor.h>?? ?? class?MyTimerHandler?:?public?ACE_Event_Handler?? {?? private:?? ????int?inteval;????//執行時間間隔?? ????int?delay;????????//延遲執行時間?? ????int?timerid;?? ?? public:?? ????MyTimerHandler(int?delay,int?inteval)?? ????{?? ????????this->delay=delay;?? ????????this->inteval=inteval;?? ????}?? ?? ????int?open()????//注冊定時器?? ????{?? ????????ACE_Time_Value?delaytime(inteval);?? ????????ACE_Time_Value?intevaltime(inteval);?? ????????timerid?=?reactor()->schedule_timer(this,?? ????????????0,????//傳遞handle_timeout給的參數?? ????????????delaytime,?? ????????????intevaltime);?? ????????return?timerid;?? ????}?? ?? ????int?close()????//取消定時器?? ????{?? ????????return?reactor()->cancel_timer(timerid);?? ????}?? ?? ????//定時器回調函數?? ????int?handle_timeout?(const?ACE_Time_Value?¤t_time,?? ????????const?void?*?=?0)?? ????{?? ????????time_t?epoch?=?((timespec_t)current_time).tv_sec;?? ????????ACE_DEBUG?((LM_INFO,?? ????????????ACE_TEXT?("handle_timeout:?%s\n"),?? ????????????ACE_OS::ctime?(&epoch)));?? ????????return?0;?? ????}?? };?? ?? int?main(int?argc,?char?*argv[])??? {?? ????MyTimerHandler?*?timer?=?new?MyTimerHandler?(3,5);?? ????timer->reactor(ACE_Reactor::instance());?? ????timer->open();?? ?? ????for(int?i=0;i<2;i++)????//觸發次handle_timeout事件?? ????{?? ????????ACE_OS::printf("\n%d\n",i);?? ????????ACE_Reactor::instance()->handle_events();?? ????}?? ?? ????timer->close();?? ????ACE_OS::printf("cancel?timer");?? ????while(true)?? ????????ACE_Reactor::instance()->handle_events();?? ????return?0;??? }?? ? ?
代碼功能比較簡單，這里就不多做介紹了。
總結
以上是生活随笔為你收集整理的ACE反应器(Reactor)模式的全部內容，希望文章能夠幫你解決所遇到的問題。
如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。
