本文转自：http://senlinzhan.github.io/2017/09/17/boost-asio/

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，封装了线程池的创建操作 ：

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::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
     
       lock(mtx);                      std::cout << "Hello, World! " << std::endl;                  });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()方法 ：

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_;};
       
      
     

参考资料