Boost.Asio 有两种支持多线程的方式,第一种方式比较简单:在多线程的场景下,每个线程都持有一个io_service
,并且每个线程都调用各自的io_service
的run()
方法。
另一种支持多线程的方式:全局只分配一个io_service
,并且让这个io_service
在多个线程之间共享,每个线程都调用全局的io_service
的run()
方法。
每个线程一个 I/O Service
让我们先分析第一种方案:在多线程的场景下,每个线程都持有一个io_service
(通常的做法是,让线程数和 CPU 核心数保持一致)。那么这种方案有什么特点呢?
- 在多核的机器上,这种方案可以充分利用多个 CPU 核心。
- 某个 socket 描述符并不会在多个线程之间共享,所以不需要引入同步机制。
- 在 event handler 中不能执行阻塞的操作,否则将会阻塞掉
io_service
所在的线程。
下面我们实现了一个AsioIOServicePool
,封装了线程池的创建操作:
1 class AsioIOServicePool 2 { 3 public: 4 using IOService = boost::asio::io_service; 5 using Work = boost::asio::io_service::work; 6 using WorkPtr = std::unique_ptr<Work>; 7 AsioIOServicePool(std::size_t size = std::thread::hardware_concurrency()) 8 : ioServices_(size), 9 works_(size), 10 nextIOService_(0) 11 { 12 for (std::size_t i = 0; i < size; ++i) 13 { 14 works_[i] = std::unique_ptr<Work>(new Work(ioServices_[i])); 15 } 16 for (std::size_t i = 0; i < ioServices_.size(); ++i) 17 { 18 threads_.emplace_back([this, i] () 19 { 20 ioServices_[i].run(); 21 }); 22 } 23 } 24 AsioIOServicePool(const AsioIOServicePool &) = delete; 25 AsioIOServicePool &operator=(const AsioIOServicePool &) = delete; 26 // 使用 round-robin 的方式返回一个 io_service 27 boost::asio::io_service &getIOService() 28 { 29 auto &service = ioServices_[nextIOService_++]; 30 if (nextIOService_ == ioServices_.size()) 31 { 32 nextIOService_ = 0; 33 } 34 return service; 35 } 36 void stop() 37 { 38 for (auto &work: works_) 39 { 40 work.reset(); 41 } 42 for (auto &t: threads_) 43 { 44 t.join(); 45 } 46 } 47 private: 48 std::vector<IOService> ioServices_; 49 std::vector<WorkPtr> works_; 50 std::vector<std::thread> threads_; 51 std::size_t nextIOService_; 52 };
AsioIOServicePool
使用起来也很简单:
1 std::mutex mtx; // protect std::cout 2 AsioIOServicePool pool; 3 4 boost::asio::steady_timer timer{pool.getIOService(), std::chrono::seconds{2}}; 5 timer.async_wait([&mtx] (const boost::system::error_code &ec) 6 { 7 std::lock_guard<std::mutex> lock(mtx); 8 std::cout << "Hello, World! " << std::endl; 9 }); 10 pool.stop();
一个 I/O Service 与多个线程
另一种方案则是先分配一个全局io_service
,然后开启多个线程,每个线程都调用这个io_service
的run()
方法。这样,当某个异步事件完成时,io_service
就会将相应的 event handler 交给任意一个线程去执行。
然而这种方案在实际使用中,需要注意一些问题:
- 在 event handler 中允许执行阻塞的操作 (例如数据库查询操作)。
- 线程数可以大于 CPU 核心数,譬如说,如果需要在 event handler 中执行阻塞的操作,为了提高程序的响应速度,这时就需要提高线程的数目。
- 由于多个线程同时运行事件循环(event loop),所以会导致一个问题:即一个 socket 描述符可能会在多个线程之间共享,容易出现竞态条件 (race condition)。譬如说,如果某个 socket 的可读事件很快发生了两次,那么就会出现两个线程同时读同一个 socket 的问题 (可以使用
strand
解决这个问题)。
下面实现了一个线程池,在每个 worker 线程中执行io_service
的run()
方法:
1 class AsioThreadPool 2 { 3 public: 4 AsioThreadPool(int threadNum = std::thread::hardware_concurrency()) 5 : work_(new boost::asio::io_service::work(service_)) 6 { 7 for (int i = 0; i < threadNum; ++i) 8 { 9 threads_.emplace_back([this] () { service_.run(); }); 10 } 11 } 12 AsioThreadPool(const AsioThreadPool &) = delete; 13 AsioThreadPool &operator=(const AsioThreadPool &) = delete; 14 boost::asio::io_service &getIOService() 15 { 16 return service_; 17 } 18 void stop() 19 { 20 work_.reset(); 21 for (auto &t: threads_) 22 { 23 t.join(); 24 } 25 } 26 private: 27 boost::asio::io_service service_; 28 std::unique_ptr<boost::asio::io_service::work> work_; 29 std::vector<std::thread> threads_; 30 };
无锁的同步方式
要怎样解决前面提到的竞态条件呢?Boost.Asio 提供了io_service::strand
:如果多个 event handler 通过同一个 strand 对象分发 (dispatch),那么这些 event handler 就会保证顺序地执行。
例如,下面的例子使用 strand,所以不需要使用互斥锁保证同步了 :
1 AsioThreadPool pool(4); // 开启 4 个线程 2 boost::asio::steady_timer timer1{pool.getIOService(), std::chrono::seconds{1}}; 3 boost::asio::steady_timer timer2{pool.getIOService(), std::chrono::seconds{1}}; 4 int value = 0; 5 boost::asio::io_service::strand strand{pool.getIOService()}; 6 7 timer1.async_wait(strand.wrap([&value] (const boost::system::error_code &ec) 8 { 9 std::cout << "Hello, World! " << value++ << std::endl; 10 })); 11 timer2.async_wait(strand.wrap([&value] (const boost::system::error_code &ec) 12 { 13 std::cout << "Hello, World! " << value++ << std::endl; 14 })); 15 pool.stop();
多线程 Echo Server
下面的EchoServer
可以在多线程中使用,它使用asio::strand
来解决前面提到的竞态问题:
1 class TCPConnection : public std::enable_shared_from_this<TCPConnection> 2 { 3 public: 4 TCPConnection(boost::asio::io_service &io_service) 5 : socket_(io_service), 6 strand_(io_service) 7 { } 8 9 tcp::socket &socket() { return socket_; } 10 void start() { doRead(); } 11 12 private: 13 void doRead() 14 { 15 auto self = shared_from_this(); 16 socket_.async_read_some( 17 boost::asio::buffer(buffer_, buffer_.size()), 18 strand_.wrap([this, self](boost::system::error_code ec, 19 std::size_t bytes_transferred) 20 { 21 if (!ec) { doWrite(bytes_transferred); } 22 })); 23 } 24 void doWrite(std::size_t length) 25 { 26 auto self = shared_from_this(); 27 boost::asio::async_write( 28 socket_, boost::asio::buffer(buffer_, length), 29 strand_.wrap([this, self](boost::system::error_code ec, 30 std::size_t /* bytes_transferred */) 31 { 32 if (!ec) { doRead(); } 33 })); 34 } 35 private: 36 tcp::socket socket_; 37 boost::asio::io_service::strand strand_; 38 std::array<char, 8192> buffer_; 39 }; 40 class EchoServer 41 { 42 public: 43 EchoServer(boost::asio::io_service &io_service, unsigned short port) 44 : io_service_(io_service), 45 acceptor_(io_service, tcp::endpoint(tcp::v4(), port)) 46 { 47 doAccept(); 48 } 49 void doAccept() 50 { 51 auto conn = std::make_shared<TCPConnection>(io_service_); 52 acceptor_.async_accept(conn->socket(), 53 [this, conn](boost::system::error_code ec) 54 { 55 if (!ec) { conn->start(); } 56 this->doAccept(); 57 }); 58 } 59 60 private: 61 boost::asio::io_service &io_service_; 62 tcp::acceptor acceptor_; 63 };