快速瞭解Boost.Asio 的多執行緒模型
Boost.Asio 有兩種支援多執行緒的方式,第一種方式比較簡單:在多執行緒的場景下,每個執行緒都持有一個io_service,並且每個執行緒都呼叫各自的io_service的run()方法。
另一種支援多執行緒的方式:全域性只分配一個io_service,並且讓這個io_service在多個執行緒之間共享,每個執行緒都呼叫全域性的io_service的run()方法。
每個執行緒一個 I/O Service
讓我們先分析第一種方案:在多執行緒的場景下,每個執行緒都持有一個io_service (通常的做法是,讓執行緒數和 CPU 核心數保持一致)。那麼這種方案有什麼特點呢?
1 在多核的機器上,這種方案可以充分利用多個 CPU 核心。
2 某個 socket 描述符並不會在多個執行緒之間共享,所以不需要引入同步機制。
3 在 event handler 中不能執行阻塞的操作,否則將會阻塞掉io_service所在的執行緒。
下面我們實現了一個AsioIOServicePool,封裝了執行緒池的建立操作:
class AsioIOServicePool { public: using IOService = boost::asio::io_service; using Work = boost::asio::io_service::work; using WorkPtr = std::unique_ptr<Work>; AsioIOServicePool(std::size_t size = std::thread::hardware_concurrency()) : ioServices_(size),works_(size),nextIOService_(0) { for (std::size_t i = 0; i < size; ++i) { works_[i] = std::unique_ptr<Work>(new Work(ioServices_[i])); } for (std::size_t i = 0; i < ioServices_.size(); ++i) { threads_.emplace_back([this,i] () { ioServices_[i].run(); }); } } AsioIOServicePool(const AsioIOServicePool &) = delete; AsioIOServicePool &operator=(const AsioIOServicePool &) = delete; // 使用 round-robin 的方式返回一個 io_service boost::asio::io_service &getIOService() { auto &service = ioServices_[nextIOService_++]; if (nextIOService_ == ioServices_.size()) { nextIOService_ = 0; } return service; } void stop() { for (auto &work: works_) { work.reset(); } for (auto &t: threads_) { t.join(); } } private: std::vector<IOService> ioServices_; std::vector<WorkPtr> works_; std::vector<std::thread> threads_; std::size_t nextIOService_; };
AsioIOServicePool使用起來也很簡單:
std::mutex mtx; // protect std::cout AsioIOServicePool pool; boost::asio::steady_timer timer{pool.getIOService(),std::chrono::seconds{2}}; timer.async_wait([&mtx] (const boost::system::error_code &ec) { std::lock_guard<std::mutex> lock(mtx); std::cout << "Hello,World! " << std::endl; }); pool.stop();
一個 I/O Service 與多個執行緒
另一種方案則是先分配一個全域性io_service,然後開啟多個執行緒,每個執行緒都呼叫這個io_service的run()方法。這樣,當某個非同步事件完成時,io_service就會將相應的 event handler 交給任意一個執行緒去執行。
然而這種方案在實際使用中,需要注意一些問題:
1 在 event handler 中允許執行阻塞的操作 (例如資料庫查詢操作)。
2 執行緒數可以大於 CPU 核心數,譬如說,如果需要在 event handler 中執行阻塞的操作,為了提高程式的響應速度,這時就需要提高執行緒的數目。
3 由於多個執行緒同時執行事件迴圈(event loop),所以會導致一個問題:即一個 socket 描述符可能會在多個執行緒之間共享,容易出現競態條件 (race condition)。譬如說,如果某個 socket 的可讀事件很快發生了兩次,那麼就會出現兩個執行緒同時讀同一個 socket 的問題 (可以使用strand解決這個問題)。
下面實現了一個執行緒池,在每個 worker 執行緒中執行io_service的run()方法:
class AsioThreadPool { public: AsioThreadPool(int threadNum = std::thread::hardware_concurrency()) : work_(new boost::asio::io_service::work(service_)) { for (int i = 0; i < threadNum; ++i) { threads_.emplace_back([this] () { service_.run(); }); } } AsioThreadPool(const AsioThreadPool &) = delete; AsioThreadPool &operator=(const AsioThreadPool &) = delete; boost::asio::io_service &getIOService() { return service_; } void stop() { work_.reset(); for (auto &t: threads_) { t.join(); } } private: boost::asio::io_service service_; std::unique_ptr<boost::asio::io_service::work> work_; std::vector<std::thread> threads_; };
無鎖的同步方式
要怎樣解決前面提到的競態條件呢?Boost.Asio 提供了io_service::strand:如果多個 event handler 通過同一個 strand 物件分發 (dispatch),那麼這些 event handler 就會保證順序地執行。
例如,下面的例子使用 strand,所以不需要使用互斥鎖保證同步了 :
AsioThreadPool pool(4); // 開啟 4 個執行緒 boost::asio::steady_timer timer1{pool.getIOService(),std::chrono::seconds{1}}; boost::asio::steady_timer timer2{pool.getIOService(),std::chrono::seconds{1}}; int value = 0; boost::asio::io_service::strand strand{pool.getIOService()}; timer1.async_wait(strand.wrap([&value] (const boost::system::error_code &ec) { std::cout << "Hello,World! " << value++ << std::endl; })); timer2.async_wait(strand.wrap([&value] (const boost::system::error_code &ec) { std::cout << "Hello,World! " << value++ << std::endl; })); pool.stop();
多執行緒 Echo Server
下面的EchoServer可以在多執行緒中使用,它使用asio::strand來解決前面提到的競態問題:
class TCPConnection : public std::enable_shared_from_this<TCPConnection> { public: TCPConnection(boost::asio::io_service &io_service) : socket_(io_service),strand_(io_service) { } tcp::socket &socket() { return socket_; } void start() { doRead(); } private: void doRead() { auto self = shared_from_this(); socket_.async_read_some( boost::asio::buffer(buffer_,buffer_.size()),strand_.wrap([this,self](boost::system::error_code ec,std::size_t bytes_transferred) { if (!ec) { doWrite(bytes_transferred); } })); } void doWrite(std::size_t length) { auto self = shared_from_this(); boost::asio::async_write( socket_,boost::asio::buffer(buffer_,length),std::size_t /* bytes_transferred */) { if (!ec) { doRead(); } })); } private: tcp::socket socket_; boost::asio::io_service::strand strand_; std::array<char,8192> buffer_; }; class EchoServer { public: EchoServer(boost::asio::io_service &io_service,unsigned short port) : io_service_(io_service),acceptor_(io_service,tcp::endpoint(tcp::v4(),port)) { doAccept(); } void doAccept() { auto conn = std::make_shared<TCPConnection>(io_service_); acceptor_.async_accept(conn->socket(),[this,conn](boost::system::error_code ec) { if (!ec) { conn->start(); } this->doAccept(); }); } private: boost::asio::io_service &io_service_; tcp::acceptor acceptor_; };
以上就是快速瞭解Boost.Asio 的多執行緒模型的詳細內容,更多關於c++ Boost.Asio 多執行緒模型的資料請關注我們其它相關文章!