快速了解Boost.Asio 的多線程模型

更新時(shí)間：2020年06月08日 09:10:37 作者：Boblim

這篇文章主要介紹了Boost.Asio 的多線程模型的相關(guān)知識(shí)，文中代碼非常詳細(xì)，供大家參考和學(xué)習(xí)，感興趣的朋友可以了解下

Boost.Asio 有兩種支持多線程的方式，第一種方式比較簡(jiǎn)單：在多線程的場(chǎng)景下，每個(gè)線程都持有一個(gè)io_service，并且每個(gè)線程都調(diào)用各自的io_service的run()方法。

　　另一種支持多線程的方式：全局只分配一個(gè)io_service，并且讓這個(gè)io_service在多個(gè)線程之間共享，每個(gè)線程都調(diào)用全局的io_service的run()方法。

每個(gè)線程一個(gè) I/O Service

　　讓我們先分析第一種方案：在多線程的場(chǎng)景下，每個(gè)線程都持有一個(gè)io_service (通常的做法是，讓線程數(shù)和 CPU 核心數(shù)保持一致)。那么這種方案有什么特點(diǎn)呢？

1 在多核的機(jī)器上，這種方案可以充分利用多個(gè) CPU 核心。

2 某個(gè) socket 描述符并不會(huì)在多個(gè)線程之間共享，所以不需要引入同步機(jī)制。

3 在 event handler 中不能執(zhí)行阻塞的操作，否則將會(huì)阻塞掉io_service所在的線程。

　　下面我們實(shí)現(xiàn)了一個(gè)AsioIOServicePool，封裝了線程池的創(chuàng)建操作:

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 的方式返回一個(gè) 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使用起來也很簡(jiǎn)單：

std::mutex mtx;       // protect std::cout
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();

一個(gè) I/O Service 與多個(gè)線程

　　另一種方案則是先分配一個(gè)全局io_service，然后開啟多個(gè)線程，每個(gè)線程都調(diào)用這個(gè)io_service的run()方法。這樣，當(dāng)某個(gè)異步事件完成時(shí)，io_service就會(huì)將相應(yīng)的 event handler 交給任意一個(gè)線程去執(zhí)行。

　　然而這種方案在實(shí)際使用中，需要注意一些問題：

1 在 event handler 中允許執(zhí)行阻塞的操作 (例如數(shù)據(jù)庫查詢操作)。

2 線程數(shù)可以大于 CPU 核心數(shù)，譬如說，如果需要在 event handler 中執(zhí)行阻塞的操作，為了提高程序的響應(yīng)速度，這時(shí)就需要提高線程的數(shù)目。

3 由于多個(gè)線程同時(shí)運(yùn)行事件循環(huán)(event loop)，所以會(huì)導(dǎo)致一個(gè)問題：即一個(gè) socket 描述符可能會(huì)在多個(gè)線程之間共享，容易出現(xiàn)競(jìng)態(tài)條件 (race condition)。譬如說，如果某個(gè) socket 的可讀事件很快發(fā)生了兩次，那么就會(huì)出現(xiàn)兩個(gè)線程同時(shí)讀同一個(gè) socket 的問題 (可以使用strand解決這個(gè)問題)。

　　下面實(shí)現(xiàn)了一個(gè)線程池，在每個(gè) worker 線程中執(zhí)行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<boost::asio::io_service::work> work_;
  std::vector<std::thread> threads_;
};

無鎖的同步方式

　　要怎樣解決前面提到的競(jìng)態(tài)條件呢？Boost.Asio 提供了io_service::strand：如果多個(gè) event handler 通過同一個(gè) strand 對(duì)象分發(fā) (dispatch)，那么這些 event handler 就會(huì)保證順序地執(zhí)行。

　　例如，下面的例子使用 strand，所以不需要使用互斥鎖保證同步了 :

AsioThreadPool pool(4);  // 開啟 4 個(gè)線程
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來解決前面提到的競(jìng)態(tài)問題:

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 的多線程模型的詳細(xì)內(nèi)容，更多關(guān)于c++ Boost.Asio 多線程模型的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: