ACE_Select_Reactor 一 ——入门
? ? ACE Reactor 框架實(shí)現(xiàn)了Reactor模式，允許事件驅(qū)動(dòng)的應(yīng)用對(duì)源自許多不同事件源的事件做出反映，比如IO句柄，定時(shí)器，以及信號(hào)，應(yīng)用重新定義框架所定義的掛鉤方法。框架隨機(jī)對(duì)其進(jìn)行分派來(lái)處理事件，Reactor負(fù)責(zé)：（1）檢查多路分離器來(lái)自各種事件源的、不同類型的連接和數(shù)據(jù)事件，（2）將這些事件分派給應(yīng)用所定義的處理器，由它進(jìn)行處理。
? ? 反應(yīng)式服務(wù)器響應(yīng)來(lái)自一個(gè)或多個(gè)事件源的時(shí)間，在理想情況下，對(duì)時(shí)間的響應(yīng)會(huì)足夠快，以使所有請(qǐng)求看起來(lái)像是被同時(shí)處理的，盡管事件處理通常是由單個(gè)線程處理的。同步事件多路分離器位于各個(gè)反應(yīng)式服務(wù)器的心臟處。這種機(jī)制檢測(cè)源自許多事件源的事件并對(duì)其作出響應(yīng)，從而同步地使事件作為服務(wù)器正常執(zhí)行路徑的一部分提供給服務(wù)器。
? ? select（）函數(shù)是最為常見(jiàn)的同步事件多路分離器。這個(gè)系統(tǒng)函數(shù)在同一組IO句柄上等待指定的事件發(fā)生，當(dāng)一個(gè)或者是多個(gè)IO句柄開(kāi)始活動(dòng)時(shí)，或是在指定的時(shí)間過(guò)去后，select函數(shù)就會(huì)返回。
? ? ACE_Select_Reactor是ACE_Reactor接口的一種實(shí)現(xiàn)，它使用select同步時(shí)間多路分離器函數(shù)來(lái)檢測(cè)IO和定時(shí)器事件，除了支持ACE_Reactor接口的所有特性外，ACE_Select_Reactor類還提供了以下能力：
? ? 1、它支持重入的反應(yīng)器調(diào)用，應(yīng)用可以從正在由統(tǒng)一反應(yīng)器分派的事件處理器中調(diào)用handle_event方法；
? ? 2、它可以被配置為同步化的或異步化的，在線程安全性和降低開(kāi)銷之間進(jìn)行折中；
? ? 3、它在再次調(diào)用select函數(shù)之前，分派其句柄集中的所有活動(dòng)句柄，從而保證了公正性。
?
?
[cpp] view plain copy? ? 在上面的代碼中，在51行，由ACE_Thread_Manager單體派生一個(gè)線程，并讓其運(yùn)行event_loop()函數(shù)，在52行，由ACE_Thread_Manager單體派生一個(gè)線程，讓其運(yùn)行controller()函數(shù)。
? ? 在從main函數(shù)返回之前，等待其他兩個(gè)線程推出，ACE_Thread_Manager:wait()還會(huì)收取兩個(gè)線程的退出狀態(tài)，以免內(nèi)存泄漏。
ACE_Select_Reactor 2 —— 服務(wù)器網(wǎng)關(guān)
? ? ACE中的流包裝提供面向連接的通信。流數(shù)據(jù)傳輸包裝類包括ACE_SOCK_Stream和ACE_LSOCK_Stream，他們分別包裝TCP/IP和UNIX域Socket協(xié)議數(shù)據(jù)傳輸功能。連接建立類包括針對(duì)TCP/IP的ACE_SOCK_Connector和ACE_SOCK_Acceptor，以及針對(duì)UNIX域Socket的ACE_LSOCK_Connector和ACE_LSOCK_Acceptor。
?
?
[cpp] view plain copy? ? ?在上面的Server類中，創(chuàng)建了一個(gè)被動(dòng)服務(wù)器，偵聽(tīng)到來(lái)的客戶端連接，在連接建立之后，服務(wù)器接收來(lái)自客戶端的數(shù)據(jù)，然后關(guān)閉鏈接。
? ? Server類包含的accept_connection()方法使用接收器來(lái)將連接接受“進(jìn)”ACE_SOCK_Stream new_stream_。該操作完成的基本流程是：調(diào)用接收器上的accept()，并將流作為參數(shù)傳入其中。一旦連接已經(jīng)建立進(jìn)流中，流的包裝方法send()和recv()就可以用來(lái)在新建立的鏈路上發(fā)送和接收數(shù)據(jù)，還有一個(gè)空的ACE_INET_Addr也被傳入接收器的accept()方法，并在其中被設(shè)定為發(fā)起連接的遠(yuǎn)程機(jī)器地址。
? ? 在連接建立后，服務(wù)器調(diào)用handle_connection()方法，它開(kāi)始從客戶端那里收取一個(gè)預(yù)先知道的單詞，然后將流關(guān)閉。連接關(guān)閉通過(guò)調(diào)用流上的close()方法來(lái)完成，該方法會(huì)釋放所有的Socket資源并終止連接。
?
? ? http://acme-ltt.iteye.com/blog/1455556中提到的ACE_Select_Reactor，在static?ACE_THR_FUNC_RETURN?controller?(void?*arg)函數(shù)中，調(diào)用上述的Server類，搭建基于ACE_Select_Reactor的Socket服務(wù)器網(wǎng)關(guān)。
?
?
? ? 客戶端程序：
?
[cpp] view plain copy? ? ?客戶端由單個(gè)Client類表示。Client含有connect_to_server()和send_to_server()方法。
? ? connect_to_server()方法使用類型為ACE_SOCK_Connector的連接器來(lái)主動(dòng)地建立連接。連接的設(shè)置通過(guò)調(diào)用連接器上的connect()方法來(lái)完成：傳入的參數(shù)為想要連接的機(jī)器的遠(yuǎn)程地址，以及用于在其中建立連接的空ACE_SOCK_Stream 。遠(yuǎn)程機(jī)器在運(yùn)行時(shí)參數(shù)中指定。一旦connect()方法成功返回，通過(guò)使用ACE_SOCK_Stream封裝類中的send()和recv()方法族，流就可以用于在新建立的鏈路上發(fā)送和接收數(shù)據(jù)。
?
? ? 在該代碼中，一旦連接建立好，send_to_server()方法就會(huì)被調(diào)用，將一個(gè)iovec類型的數(shù)組，用sendv_n()方法，發(fā)送到服務(wù)器上。
采用 ACE Reactor 實(shí)現(xiàn)服務(wù)程序例子
此文版權(quán)屬于作者所有，任何人、媒體或者網(wǎng)站轉(zhuǎn)載、借用都必須征得作者本人同意！
ACE 使用方法及例子，網(wǎng)上有不少，下面貼一段我寫的采用 ACE Reactor 模式寫的 echo 服務(wù)的例子代碼，通過(guò)例子可以看出，采用 ACE 開(kāi)發(fā)多客戶端的服務(wù)程序那是相當(dāng)簡(jiǎn)單的！
代碼中，handle_input(…)和 handle_output(…)都會(huì)對(duì) _bufs?進(jìn)行操作，因?yàn)檫@兩個(gè)函數(shù)都是運(yùn)行在 reactor 的線程里，不會(huì)沖突，所以沒(méi)有必要對(duì) _bufs 的操作進(jìn)行鎖操作。
/* $Id: cpp.tpl 3412 2009-11-14 14:23:44Z luozhiyong $ */ /** * \file ACEReactorSvrSample.cpp * * \brief?采用ACE Reactor 實(shí)現(xiàn)服務(wù)程序例子 * * \version $Rev: 3412 $ * \author?? * \date???? 2009年09月08日08:17:10 * * \note?修改歷史：<br> * <table> *???? <tr><th>日期</th><th>修改人</th><th>內(nèi)容</th></tr> *???? <tr><td>2009-9-8</td><td></td><td>創(chuàng)建初稿</td> *???? </tr> * </table> */ #include?<ace/Message_Block.h> #include?<ace/Svc_Handler.h> #include?<ace/SOCK_Acceptor.h> #include?<ace/Acceptor.h> #include?<ace/Select_Reactor.h> #include?<list> #include?<string> ? #ifdef?_DEBUG #??? define?ACE_RT_OPT?"d" #else #??? define ACE_RT_OPT #endif ? #if?defined_DLL #??? define?ACE_LIB_THREAD_OPT #else #??? define ACE_LIB_THREAD_OPT "s" #endif ? #pragma?comment(lib,?"ACE"ACE_LIB_THREAD_OPT?ACE_RT_OPT?".lib") class?EchoService :?public?ACE_Event_Handler { public: typedef?ACE_SOCK_STREAM?stream_type; typedef?EchoService?my_type; typedef?ACE_Acceptor<my_type,?ACE_SOCK_ACCEPTOR>?acceptor_type; EchoService() { printf("EchoService創(chuàng)建\n"); } ~EchoService() { printf("EchoService銷毀\n"); } //?響應(yīng)socket 已經(jīng)打開(kāi)，連接已經(jīng)建立事件 int?open(void*) { //?注冊(cè)讀事件 if?(reactor()->register_handler(this,ACE_Event_Handler::READ_MASK)) { //?無(wú)法注冊(cè)handler return?-1; } //?注冊(cè)寫事件 if?(reactor()->register_handler(this,ACE_Event_Handler::WRITE_MASK)) { //?無(wú)法注冊(cè)handler return?-1; } //?取消寫事件，等待有數(shù)據(jù)時(shí)喚醒 reactor()->cancel_wakeup(this,ACE_Event_Handler::WRITE_MASK); printf("EchoService已打開(kāi)\n"); return?0; } //?響應(yīng)有數(shù)據(jù)可讀事件 int?handle_input(ACE_HANDLE) { char?buf[24]; ssize_t?c?=?_peer.recv(buf,sizeof(buf) - 1); if?(c?== 0) { //?連接已經(jīng)關(guān)閉 return?-1; } _bufs.push_back(std::string(buf,c)); if?(_bufs.size() == 1) { //?緩沖區(qū)尺寸為1 說(shuō)明原來(lái)緩沖區(qū)為空，寫事件是取消的，這里喚醒它 reactor()->schedule_wakeup(this,ACE_Event_Handler::WRITE_MASK); } return?0; } //?響應(yīng)可以發(fā)送數(shù)據(jù)了事件 int?handle_output(ACE_HANDLE) { while?(!_bufs.empty()) { std::string&buf(*_bufs.begin()); char?const*??????s(buf.c_str()); char?const*const?e(s?+buf.size()); while?(s?!=e) { ssize_t?c(_peer.send(s,e?-?s)); if?(c?== -1 ||c?== 0) { //?發(fā)送不成功不論發(fā)送過(guò)程中是否發(fā)生阻塞， if?(ACE_OS::last_error() ==EWOULDBLOCK) { //?輸出緩沖區(qū)滿，無(wú)法再發(fā)送數(shù)據(jù)了（如果你還是繼續(xù)發(fā)送數(shù)據(jù)，發(fā)送會(huì)阻塞的） break; }else{ //?連接已關(guān)閉 break; } }else{ s?+=?c; } } if?(s?==e) { _bufs.pop_front(); }else{ buf?=?std::string(s,e?-?s); break; } } if?(_bufs.empty()) { //?緩沖區(qū)為空，取消寫事件監(jiān)聽(tīng) reactor()->cancel_wakeup(this,ACE_Event_Handler::WRITE_MASK); } //?不論發(fā)送是否成功都返回0，因?yàn)?#xff0c;如果發(fā)送失敗，handle_input 也會(huì)發(fā)生讀失敗事件， //?錯(cuò)誤處理有handle_input 返回-1 來(lái)觸發(fā) return?0; } int?handle_close(ACE_HANDLE?=?ACE_INVALID_HANDLE,?ACE_Reactor_Mask?mask?=ACE_Event_Handler::ALL_EVENTS_MASK) { if?(mask?==ACE_Event_Handler::WRITE_MASK) return?0; _peer.close(); delete?this; return?0; } //?這個(gè)函數(shù)主要給reactor::register_handler 時(shí)使用的 ACE_HANDLE?get_handle?()?const { return?_peer.get_handle(); } //?這個(gè)函數(shù)主要給acceptor 使用的 stream_type&?peer() { return?_peer; } //?這個(gè)函數(shù)主要給acceptor 使用的 int?close?(u_long?= 0) { return?handle_close(); } private: stream_type?_peer; std::list<std::string>_bufs; }; int?main(int?/*argc*/,char*?/*argv*/[]) { u_short?port?= 20001; ACE_Reactor::instance(newACE_Reactor(newACE_Select_Reactor,?true)); EchoService::acceptor_typeacceptor; ACE_INET_Addr?svrAddr(port); if?(acceptor.open(svrAddr)) { fprintf(stderr,"服務(wù)打開(kāi)失敗：%s\n",ACE_OS::strerror(ACE_OS::last_error())); return?1; }else{ fprintf(stdout,"服務(wù)已打開(kāi)，端口為：%u\n",port); ACE_Reactor::instance()->run_reactor_event_loop(); return?0; } }總結(jié)
以上是生活随笔為你收集整理的ACE_Select_Reactor 一 ——入门的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。
- 上一篇： 高性能I/O设计模式Reactor和Pr
- 下一篇： ACE_Reactor（二）ACE_De