<转>浅谈 Boost.Asio 的多线程模型-阿里云开发者社区

开发者社区> 狂斩一条龙> 正文

<转>浅谈 Boost.Asio 的多线程模型

简介: 本文转自:http://senlinzhan.github.io/2017/09/17/boost-asio/ Boost.Asio 有两种支持多线程的方式,第一种方式比较简单:在多线程的场景下,每个线程都持有一个io_service,并且每个线程都调用各自的io_service的run()方法。
+关注继续查看

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

Boost.Asio 有两种支持多线程的方式,第一种方式比较简单:在多线程的场景下,每个线程都持有一个io_service,并且每个线程都调用各自的io_servicerun()方法。
  另一种支持多线程的方式:全局只分配一个io_service,并且让这个io_service在多个线程之间共享,每个线程都调用全局的io_servicerun()方法。

每个线程一个 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<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使用起来也很简单:

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_servicerun()方法。这样,当某个异步事件完成时,io_service就会将相应的 event handler 交给任意一个线程去执行。
  然而这种方案在实际使用中,需要注意一些问题:

  • 在 event handler 中允许执行阻塞的操作 (例如数据库查询操作)。
  • 线程数可以大于 CPU 核心数,譬如说,如果需要在 event handler 中执行阻塞的操作,为了提高程序的响应速度,这时就需要提高线程的数目。
  • 由于多个线程同时运行事件循环(event loop),所以会导致一个问题:即一个 socket 描述符可能会在多个线程之间共享,容易出现竞态条件 (race condition)。譬如说,如果某个 socket 的可读事件很快发生了两次,那么就会出现两个线程同时读同一个 socket 的问题 (可以使用strand解决这个问题)。

  下面实现了一个线程池,在每个 worker 线程中执行io_servicerun()方法 [完整代码]

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,所以不需要使用互斥锁保证同步了 [完整代码]

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

参考资料

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
阿里云服务器怎么设置密码?怎么停机?怎么重启服务器?
如果在创建实例时没有设置密码,或者密码丢失,您可以在控制台上重新设置实例的登录密码。本文仅描述如何在 ECS 管理控制台上修改实例登录密码。
9496 0
使用SSH远程登录阿里云ECS服务器
远程连接服务器以及配置环境
2464 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,阿里云优惠总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系.
13185 0
windows server 2008阿里云ECS服务器安全设置
最近我们Sinesafe安全公司在为客户使用阿里云ecs服务器做安全的过程中,发现服务器基础安全性都没有做。为了为站长们提供更加有效的安全基础解决方案,我们Sinesafe将对阿里云服务器win2008 系统进行基础安全部署实战过程! 比较重要的几部分 1.
9054 0
阿里云ECS云服务器初始化设置教程方法
阿里云ECS云服务器初始化是指将云服务器系统恢复到最初状态的过程,阿里云的服务器初始化是通过更换系统盘来实现的,是免费的,阿里云百科网分享服务器初始化教程: 服务器初始化教程方法 本文的服务器初始化是指将ECS云服务器系统恢复到最初状态,服务器中的数据也会被清空,所以初始化之前一定要先备份好。
6893 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,云吞铺子总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系统盘、创建快照、配置安全组等操作如何登录ECS云服务器控制台? 1、先登录到阿里云ECS服务器控制台 2、点击顶部的“控制台” 3、通过左侧栏,切换到“云服务器ECS”即可,如下图所示 通过ECS控制台的远程连接来登录到云服务器 阿里云ECS云服务器自带远程连接功能,使用该功能可以登录到云服务器,简单且方便,如下图:点击“远程连接”,第一次连接会自动生成6位数字密码,输入密码即可登录到云服务器上。
21908 0
阿里云服务器ECS登录用户名是什么?系统不同默认账号也不同
阿里云服务器Windows系统默认用户名administrator,Linux镜像服务器用户名root
4012 0
+关注
狂斩一条龙
惯用c++,在windows平台做应用开发和底层研究,并对ua语言有深入的了解
476
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载