Boost.Asio 有两种支持多线程的方式，第一种方式比较简单：在多线程的场景下，每个线程都持有一个 io_service，并且每个线程都调用各自的 io_service 的 run() 方法（io_service 的运行机制可以看本站的《深入理解 Boost.Asio 的 io_service》）。
另一种支持多线程的方式：全局只分配一个 io_service，并且让这个 io_service 在多个线程之间共享，每个线程都调用全局的 io_service 的 run() 方法。
每个线程一个 I/O Service
让我们先分析第一种方案：在多线程的场景下，每个线程都持有一个 io_service（通常的做法是，让线程数和 CPU 核心数保持一致）。这样，每个 io_service 都由独立的线程驱动，彼此之间互不竞争；代价是当某个 io_service 暂时没有 I/O 事件时，它对应的线程就只能空闲等待。下面我们实现了一个 AsioIOServicePool，封装了线程池的创建操作，完整代码如下：
#include <cstddef>
#include <memory>
#include <thread>
#include <vector>
#include <boost/asio.hpp>
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 使用起来也很简单：
#include <chrono>
#include <iostream>
#include <mutex>
#include <boost/asio.hpp>
std::mutex mtx; // 保护 std::cout
int main() {
// 使用上文定义的 AsioIOServicePool
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();
return 0;
}一个 I/O Service 与多个线程
另一种方案则是先分配一个全局 io_service，然后开启多个线程，每个线程都调用这个 io_service 的 run() 方法。这样，当某个异步事件完成时，io_service 就会将相应的 event handler 交给任意一个线程去执行。
然而这种方案在实际使用中，需要注意一些问题：由于同一个 io_service 的事件处理器可能在多个线程中并发执行，因此会产生竞态条件。
下面实现了一个线程池，在每个 worker 线程中执行 io_service 的 run() 方法，完整代码如下：
#include <memory>
#include <thread>
#include <vector>
#include <boost/asio.hpp>
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，所以不需要使用互斥锁保证同步了，完整代码如下：
#include <chrono>
#include <iostream>
#include <boost/asio.hpp>
int main() {
// 使用上文定义的 AsioThreadPool
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();
return 0;
}多线程 Echo Server
下面的 EchoServer 可以在多线程中使用，它使用 io_service::strand 来解决前面提到的竞态问题，完整代码如下：
#include <array>
#include <iostream>
#include <memory>
#include <boost/asio.hpp>
using boost::asio::ip::tcp;
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_;
};
// 复用上文的 AsioThreadPool：多个线程共享同一个 io_service
int main() {
AsioThreadPool pool;
EchoServer server(pool.getIOService(), 8888);
std::cin.get(); // 阻塞主线程，保持服务运行
pool.stop();
return 0;
}关于单线程版本的回显服务端实现，可以对照本站的《Boost.Asio：客户端和服务端》。
