浅谈 Boost.Asio 的多线程模型

每个线程一个 I/O Service

让我们先分析第一种方案：在多线程的场景下，每个线程都持有一个 io_service (通常的做法是，让线程数和 CPU 核心数保持一致)。那么这种方案有什么特点呢？

在多核的机器上，这种方案可以充分利用多个 CPU 核心。

某个 socket 描述符并不会在多个线程之间共享，所以不需要引入同步机制。

在 event handler 中不能执行阻塞的操作，否则将会阻塞掉


    io_service

所在的线程。

下面我们实现了一个 AsioIOServicePool ，封装了线程池的创建操作:

 1 class AsioIOServicePool
 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;
 4 boost::asio::steady_timer timer{pool.getIOService(), std::chrono::seconds{2}};
 5 timer.async_wait([&mtx] (const boost::system::error_code &ec)
 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
 3 public:
 4     AsioThreadPool(int threadNum = std::thread::hardware_concurrency())
 5         : work_(new boost::asio::io_service::work(service_))
 7         for (int i = 0; i < threadNum; ++i)
 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()};    
 7 timer1.async_wait(strand.wrap([&value] (const boost::system::error_code &ec)
 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> 
 3 public:
 4     TCPConnection(boost::asio::io_service &io_service)
 5         : socket_(io_service),
 6           strand_(io_service)
 7     { }
 9     tcp::socket &socket() {  return socket_;  }
10     void start() {  doRead();  }
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     }
60 private: 
61     boost::asio::io_service &io_service_;
62     tcp::acceptor acceptor_;
63 };