快速了解Boost.Asio 的多線程模型
Boost.Asio 有兩種支持多線程的方式，第一種方式比較簡(jiǎn)單：在多線程的場(chǎng)景下，每個(gè)線程都持有一個(gè)io_service，并且每個(gè)線程都調(diào)用各自的io_service的run()方法。
另一種支持多線程的方式：全局只分配一個(gè)io_service，并且讓這個(gè)io_service在多個(gè)線程之間共享，每個(gè)線程都調(diào)用全局的io_service的run()方法。
每個(gè)線程一個(gè) I/O Service
讓我們先分析第一種方案：在多線程的場(chǎng)景下，每個(gè)線程都持有一個(gè)io_service (通常的做法是，讓線程數(shù)和 CPU 核心數(shù)保持一致)。那么這種方案有什么特點(diǎn)呢？
1 在多核的機(jī)器上，這種方案可以充分利用多個(gè) CPU 核心。
2 某個(gè) socket 描述符并不會(huì)在多個(gè)線程之間共享，所以不需要引入同步機(jī)制。
3 在 event handler 中不能執(zhí)行阻塞的操作，否則將會(huì)阻塞掉io_service所在的線程。
下面我們實(shí)現(xiàn)了一個(gè)AsioIOServicePool，封裝了線程池的創(chuàng)建操作:
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 的方式返回一個(gè) 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使用起來(lái)也很簡(jiǎn)單：
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();
一個(gè) I/O Service 與多個(gè)線程
另一種方案則是先分配一個(gè)全局io_service，然后開(kāi)啟多個(gè)線程，每個(gè)線程都調(diào)用這個(gè)io_service的run()方法。這樣，當(dāng)某個(gè)異步事件完成時(shí)，io_service就會(huì)將相應(yīng)的 event handler 交給任意一個(gè)線程去執(zhí)行。
然而這種方案在實(shí)際使用中，需要注意一些問(wèn)題：
1 在 event handler 中允許執(zhí)行阻塞的操作 (例如數(shù)據(jù)庫(kù)查詢操作)。
2 線程數(shù)可以大于 CPU 核心數(shù)，譬如說(shuō)，如果需要在 event handler 中執(zhí)行阻塞的操作，為了提高程序的響應(yīng)速度，這時(shí)就需要提高線程的數(shù)目。
3 由于多個(gè)線程同時(shí)運(yùn)行事件循環(huán)(event loop)，所以會(huì)導(dǎo)致一個(gè)問(wèn)題：即一個(gè) socket 描述符可能會(huì)在多個(gè)線程之間共享，容易出現(xiàn)競(jìng)態(tài)條件 (race condition)。譬如說(shuō)，如果某個(gè) socket 的可讀事件很快發(fā)生了兩次，那么就會(huì)出現(xiàn)兩個(gè)線程同時(shí)讀同一個(gè) socket 的問(wèn)題 (可以使用strand解決這個(gè)問(wèn)題)。
下面實(shí)現(xiàn)了一個(gè)線程池，在每個(gè) worker 線程中執(zhí)行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_;
};
無(wú)鎖的同步方式
要怎樣解決前面提到的競(jìng)態(tài)條件呢？Boost.Asio 提供了io_service::strand：如果多個(gè) event handler 通過(guò)同一個(gè) strand 對(duì)象分發(fā) (dispatch)，那么這些 event handler 就會(huì)保證順序地執(zhí)行。
例如，下面的例子使用 strand，所以不需要使用互斥鎖保證同步了 :
AsioThreadPool pool(4); // 開(kāi)啟 4 個(gè)線程
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來(lái)解決前面提到的競(jìng)態(tài)問(wèn)題:
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),
strand_.wrap([this, self](boost::system::error_code ec,
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 的多線程模型的詳細(xì)內(nèi)容，更多關(guān)于c++ Boost.Asio 多線程模型的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
詳解C++中的指針結(jié)構(gòu)體數(shù)組以及指向結(jié)構(gòu)體變量的指針
這篇文章主要介紹了C++中的指針結(jié)構(gòu)體數(shù)組以及指向結(jié)構(gòu)體變量的指針的用法,是C++入門學(xué)習(xí)中的基礎(chǔ)知識(shí),需要的朋友可以參考下2015-09-09
今年最火爆的解壓小游戲電子木魚(yú)，現(xiàn)在許多軟件都上架了這個(gè)小程序。我在網(wǎng)上看了一下基本上都是用py和Java寫(xiě)的，所以我用QT重新寫(xiě)了一下，作為小白練手項(xiàng)目非常適合，快跟隨小編一起學(xué)習(xí)一下吧2023-01-01
這篇文章主要介紹了c++中system("pause")的作用和含義，非常不錯(cuò)，具有參考借鑒價(jià)值，需要的朋友參考下吧2018-03-03
這篇文章主要介紹了C語(yǔ)言中二級(jí)指針的實(shí)例詳解的相關(guān)資料,希望通過(guò)本文能幫助到大家，讓大家掌握理解二級(jí)指針的知識(shí)，需要的朋友可以參考下2017-10-10
在C++中把字符串轉(zhuǎn)換為整數(shù)的兩種簡(jiǎn)單方法
經(jīng)常會(huì)遇到類型轉(zhuǎn)換，本文主要介紹了C++中把字符串轉(zhuǎn)換為整數(shù)的兩種簡(jiǎn)單方法，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2022-06-06
C++日期類(Date)實(shí)現(xiàn)的示例代碼
這篇文章主要為大家詳細(xì)介紹了如何利用C++語(yǔ)言實(shí)現(xiàn)日期類(Date)，可以實(shí)現(xiàn)確定某年某月有多少天、打印日期等功能，感興趣的可以了解一下2022-07-07
一文帶你了解C++中的右值引用與移動(dòng)語(yǔ)義
本篇文章主要為大家詳細(xì)介紹了C++中的右值引用與移動(dòng)語(yǔ)義的相關(guān)知識(shí)，小編覺(jué)得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2023-03-03
ImGUI 它是與平臺(tái)無(wú)關(guān)的C++輕量級(jí)跨平臺(tái)圖形界面庫(kù)，沒(méi)有任何第三方依賴，可以將ImGUI的源碼直接加到項(xiàng)目中使用。本文將利用ImGUI繪制D3D外部菜單，需要的可以參考一下2022-09-09
C++實(shí)現(xiàn)leetcode(3.最長(zhǎng)無(wú)重復(fù)字符的子串)
這篇文章主要介紹了C++實(shí)現(xiàn)leetcode(3.最長(zhǎng)無(wú)重復(fù)字符的子串),本篇文章通過(guò)簡(jiǎn)要的案例,講解了該項(xiàng)技術(shù)的了解與使用,以下就是詳細(xì)內(nèi)容,需要的朋友可以參考下2021-07-07
這篇文章主要為大家詳細(xì)介紹了opencv實(shí)現(xiàn)圖形輪廓檢測(cè)，文中示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2021-04-04