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，封装了线程池的创建操作:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
|
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使用起来也很简单：
|
1
2
3
4
5
6
7
8
9
10
|
std::mutex mtx; // protect std::coutAsioIOServicePool 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()方法：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
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，所以不需要使用互斥锁保证同步了 :
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
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来解决前面提到的竞态问题:
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
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 多线程模型的资料请关注服务器之家其它相关文章！
原文链接：https://www.cnblogs.com/fnlingnzb-learner/p/10402276.html
