服务器之家

服务器之家 > 正文

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

时间:2021-09-09 15:36     来源/作者:Boblim

Boost.Asio 有两种支持多线程的方式,第一种方式比较简单:在多线程的场景下,每个线程都持有一个io_service,并且每个线程都调用各自的io_service的run()方法。

  另一种支持多线程的方式:全局只分配一个io_service,并且让这个io_service在多个线程之间共享,每个线程都调用全局的io_service的run()方法。

每个线程一个 I/O Service

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

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

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

3  在 event handler 中不能执行阻塞的操作,否则将会阻塞掉io_service所在的线程。

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

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
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使用起来也很简单:

?
1
2
3
4
5
6
7
8
9
10
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();

一个 I/O Service 与多个线程

  另一种方案则是先分配一个全局io_service,然后开启多个线程,每个线程都调用这个io_service的run()方法。这样,当某个异步事件完成时,io_service就会将相应的 event handler 交给任意一个线程去执行。

  然而这种方案在实际使用中,需要注意一些问题:

1  在 event handler 中允许执行阻塞的操作 (例如数据库查询操作)。

2  线程数可以大于 CPU 核心数,譬如说,如果需要在 event handler 中执行阻塞的操作,为了提高程序的响应速度,这时就需要提高线程的数目。

3  由于多个线程同时运行事件循环(event loop),所以会导致一个问题:即一个 socket 描述符可能会在多个线程之间共享,容易出现竞态条件 (race condition)。譬如说,如果某个 socket 的可读事件很快发生了两次,那么就会出现两个线程同时读同一个 socket 的问题 (可以使用strand解决这个问题)。

  下面实现了一个线程池,在每个 worker 线程中执行io_service的run()方法:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
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,所以不需要使用互斥锁保证同步了 :

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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来解决前面提到的竞态问题:

?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
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 的多线程模型的详细内容,更多关于c++ Boost.Asio 多线程模型的资料请关注服务器之家其它相关文章!

原文链接:https://www.cnblogs.com/fnlingnzb-learner/p/10402276.html

相关文章

热门资讯

yue是什么意思 网络流行语yue了是什么梗
yue是什么意思 网络流行语yue了是什么梗 2020-10-11
2020微信伤感网名听哭了 让对方看到心疼的伤感网名大全
2020微信伤感网名听哭了 让对方看到心疼的伤感网名大全 2019-12-26
背刺什么意思 网络词语背刺是什么梗
背刺什么意思 网络词语背刺是什么梗 2020-05-22
苹果12mini价格表官网报价 iPhone12mini全版本价格汇总
苹果12mini价格表官网报价 iPhone12mini全版本价格汇总 2020-11-13
2021年耽改剧名单 2021要播出的59部耽改剧列表
2021年耽改剧名单 2021要播出的59部耽改剧列表 2021-03-05
返回顶部