快速了解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),
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 的多线程模型的详细内容，更多关于c++ Boost.Asio 多线程模型的资料请关注脚本之家其它相关文章！
相关文章
这篇文章主要介绍了C++中的指针结构体数组以及指向结构体变量的指针的用法,是C++入门学习中的基础知识,需要的朋友可以参考下2015-09-09
今年最火爆的解压小游戏电子木鱼，现在许多软件都上架了这个小程序。我在网上看了一下基本上都是用py和Java写的，所以我用QT重新写了一下，作为小白练手项目非常适合，快跟随小编一起学习一下吧2023-01-01
这篇文章主要介绍了c++中system("pause")的作用和含义，非常不错，具有参考借鉴价值，需要的朋友参考下吧2018-03-03
这篇文章主要介绍了C语言中二级指针的实例详解的相关资料,希望通过本文能帮助到大家，让大家掌握理解二级指针的知识，需要的朋友可以参考下2017-10-10
经常会遇到类型转换，本文主要介绍了C++中把字符串转换为整数的两种简单方法，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧2022-06-06
这篇文章主要为大家详细介绍了如何利用C++语言实现日期类(Date)，可以实现确定某年某月有多少天、打印日期等功能，感兴趣的可以了解一下2022-07-07
本篇文章主要为大家详细介绍了C++中的右值引用与移动语义的相关知识，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧2023-03-03
ImGUI 它是与平台无关的C++轻量级跨平台图形界面库，没有任何第三方依赖，可以将ImGUI的源码直接加到项目中使用。本文将利用ImGUI绘制D3D外部菜单，需要的可以参考一下2022-09-09
这篇文章主要介绍了C++实现leetcode(3.最长无重复字符的子串),本篇文章通过简要的案例,讲解了该项技术的了解与使用,以下就是详细内容,需要的朋友可以参考下2021-07-07
这篇文章主要为大家详细介绍了opencv实现图形轮廓检测，文中示例代码介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们可以参考一下2021-04-04
最新评论