这篇文章主要是浅谈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;
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(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 ioServices_;
std::vector works_;
std::vector 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 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 work_;
std::vector 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{ 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 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 (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的多线程模型有进一步的了解,如果还想学习更多内容,欢迎关注创新互联行业资讯频道。
本文名称:浅谈Boost.Asio的多线程模型
转载源于:http://www.jxjierui.cn/article/iidoes.html


咨询
建站咨询
