網路方面用的比較多的庫是libevent和boost.asio,兩者都是跨平臺的。其中libevent是基於Reactor實現的,而boost.asio是基於Proactor實現的。Reactor和Proactor模式的主要區別就是真正的操作(如讀/寫)是由誰來完成的,Reactor中需要應用程式自 ...
網路方面用的比較多的庫是libevent和boost.asio,兩者都是跨平臺的。其中libevent是基於Reactor實現的,而boost.asio是基於Proactor實現的。Reactor和Proactor模式的主要區別就是真正的操作(如讀/寫)是由誰來完成的,Reactor中需要應用程式自己讀取或者寫入數據,而在Proactor模式中,應用程式不需要進行實際的讀/寫過程,操作系統會讀取緩衝區或者寫入緩衝區到真正的IO設備,應用程式只需要從緩衝區讀取(操作系統已經幫我們讀好了)或者寫入(操作系統會幫我們寫入)即可。在Proactor模式中,用戶發起非同步操作之後就返回了,讓操作系統去處理請求,然後等著回調到完成事件函數中處理非同步操作的結果。
1. 反應器(Reactor)
Reactor一般是應用程式先註冊響應的事件處理器,然後啟動Reactor的事件迴圈,不斷的檢查是否有就緒的IO事件,當有就緒IO事件發生時,反應器的事件迴圈就會調用事先註冊好的事件處理器。下麵代碼是libevent的一個簡單應用代碼及就緒的IO事件發生時的堆棧圖,其中就緒IO事件可以使用網路調試助手,連接本機之後即可產生。
#include "stdafx.h" #include <string.h> #include <errno.h> #include <stdio.h> #include <signal.h> #ifndef WIN32 #include <netinet/in.h> # ifdef _XOPEN_SOURCE_EXTENDED # include <arpa/inet.h> # endif #include <sys/socket.h> #endif #include <event2/bufferevent.h> #include <event2/buffer.h> #include <event2/listener.h> #include <event2/util.h> #include <event2/event.h> static const char MESSAGE[] = "Hello, World!\n"; static const int PORT = 9995; static void listener_cb(struct evconnlistener *, evutil_socket_t, struct sockaddr *, int socklen, void *); static void conn_writecb(struct bufferevent *, void *); static void conn_eventcb(struct bufferevent *, short, void *); static void signal_cb(evutil_socket_t, short, void *); int main(int argc, char **argv) { struct event_base *base; struct evconnlistener *listener; struct event *signal_event; struct sockaddr_in sin; #ifdef WIN32 WSADATA wsa_data; WSAStartup(0x0201, &wsa_data); #endif base = event_base_new(); if (!base) { fprintf(stderr, "Could not initialize libevent!\n"); return 1; } memset(&sin, 0, sizeof(sin)); sin.sin_family = AF_INET; sin.sin_port = htons(PORT); listener = evconnlistener_new_bind(base, listener_cb, (void *)base, LEV_OPT_REUSEABLE|LEV_OPT_CLOSE_ON_FREE, -1, (struct sockaddr*)&sin, sizeof(sin)); if (!listener) { fprintf(stderr, "Could not create a listener!\n"); return 1; } signal_event = evsignal_new(base, SIGINT, signal_cb, (void *)base); if (!signal_event || event_add(signal_event, NULL)<0) { fprintf(stderr, "Could not create/add a signal event!\n"); return 1; } event_base_dispatch(base); evconnlistener_free(listener); event_free(signal_event); event_base_free(base); printf("done\n"); return 0; } static void listener_cb(struct evconnlistener *listener, evutil_socket_t fd, struct sockaddr *sa, int socklen, void *user_data) { struct event_base *base = (event_base *)user_data; struct bufferevent *bev; bev = bufferevent_socket_new(base, fd, BEV_OPT_CLOSE_ON_FREE); if (!bev) { fprintf(stderr, "Error constructing bufferevent!"); event_base_loopbreak(base); return; } bufferevent_setcb(bev, NULL, conn_writecb, conn_eventcb, NULL); bufferevent_enable(bev, EV_WRITE); bufferevent_disable(bev, EV_READ); bufferevent_write(bev, MESSAGE, strlen(MESSAGE)); } static void conn_writecb(struct bufferevent *bev, void *user_data) { struct evbuffer *output = bufferevent_get_output(bev); if (evbuffer_get_length(output) == 0) { printf("flushed answer\n"); bufferevent_free(bev); } } static void conn_eventcb(struct bufferevent *bev, short events, void *user_data) { if (events & BEV_EVENT_EOF) { printf("Connection closed.\n"); } else if (events & BEV_EVENT_ERROR) { printf("Got an error on the connection: %s\n", strerror(errno));/*XXX win32*/ } /* None of the other events can happen here, since we haven't enabled * timeouts */ bufferevent_free(bev); } static void signal_cb(evutil_socket_t sig, short events, void *user_data) { struct event_base *base = (event_base *)user_data; struct timeval delay = { 2, 0 }; printf("Caught an interrupt signal; exiting cleanly in two seconds.\n"); event_base_loopexit(base, &delay); }
有連接時的堆棧圖:
從堆棧圖中可以看出libevent只有一個線程在執行,都是從event_base_dispatch中逐漸回調的。反應器逆置了事件的處理流程,但是可以看出它不能同時支持大量客戶請求或者耗時過長的請求,因為它串列化了所有的事件處理流程。
2. 主動器(Proactor)
(1)Proactor需要調用者定義一個非同步執行的操作,例如,socket的非同步讀/寫;
(2)執行非同步操作,非同步事件處理器將非同步請求交給操作系統就返回了,讓操作系統去完成具體的操作,操作系統在完成操作之後,會將完成事件放入一個完成事件隊列。
(3)非同步事件分離器會檢測完成事件,若檢測到完成事件,則從完成隊列中取出完成事件,並通知應用程式註冊的完成事件處理函數去處理;
(4)完成事件處理函數處理非同步操作的結果。
下麵是一個基於boost::asio的非同步伺服器:
#include "stdafx.h" #include <boost/asio.hpp> #include <boost/bind/placeholders.hpp> #include <boost/bind/bind.hpp> #include <boost/system/error_code.hpp> #include <boost/smart_ptr/enable_shared_from_this.hpp> using namespace boost::asio; namespace { typedef boost::asio::io_service IoService; typedef boost::asio::ip::tcp TCP; std::string make_daytime_string() { using namespace std; time_t now = std::time(NULL); return ctime(&now); } class tcp_connection : public boost::enable_shared_from_this<tcp_connection> { public: typedef boost::shared_ptr<tcp_connection> pointer; static pointer create(IoService& io_service) { return pointer(new tcp_connection(io_service)); } TCP::socket& socket() { return socket_; } void start() { message_ = make_daytime_string(); boost::asio::async_write( socket_, boost::asio::buffer(message_), boost::bind(&tcp_connection::handle_write, shared_from_this(), boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)); } private: tcp_connection(IoService& io_service) : socket_(io_service) { } void handle_write(const boost::system::error_code& /*error*/, size_t /*bytes_transferred*/) { printf("write data!!!"); } TCP::socket socket_; std::string message_; }; class tcp_server { public: tcp_server(IoService& io_service) : acceptor_(io_service, TCP::endpoint(TCP::v4(), 10000)) { start_accept(); } private: void start_accept() { tcp_connection::pointer new_connection = tcp_connection::create(acceptor_.get_io_service()); acceptor_.async_accept( new_connection->socket(), boost::bind(&tcp_server::handle_accept, this, new_connection, boost::asio::placeholders::error)); } void handle_accept(tcp_connection::pointer new_connection, const boost::system::error_code& error) { if (!error) { new_connection->start(); start_accept(); } } TCP::acceptor acceptor_; }; } // tcp_connection與tcp_server封裝後 void test_asio_asynserver() { try { IoService io_service; tcp_server server(io_service); // 只有io_service類的run()方法運行之後回調對象才會被調用 io_service.run(); } catch (std::exception& e) { std::cerr << e.what() << std::endl; } } int main() { test_asio_asynserver(); return 0; }
有連接時需要寫入數據,但是寫入數據並不是由用戶寫入的,而是把需要寫入的數據提交給了系統,由系統擇機寫入,堆棧如下:
總結兩者,可以看出Reactor採用的是同步IO,主動器採用的是非同步IO,同步和非同步之分可以參考文章( IO - 同步,非同步,阻塞,非阻塞 (亡羊補牢篇)),個人認為簡單來說,同步IO是發出了請求,不管阻塞還是非阻塞,都需要調用者主動去check調用的結果;而非同步IO是由被調用者通知調用者來處理結果。
參考資料:boost庫asio詳解8——幾個TCP的簡單例子