本文转自: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<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_;
    
  • The Boost C++ Libraries Chapter 32. Boost.Asio
  • A guide to getting started with boost::asio
  • Strands: Use Threads Without Explicit Locking
  • Post on ASIO strand
  • How strands guarantee correct execution of pending events in boost.asio
  • asio C++ library
  • Using Asio with C++11
  • 最近尝试使用了一下Boost.Asio,不知道是否因为各大公司都有自己相对成熟的网络库的缘故,网络上Asio相关的资料实在不多,而且很多翻来覆去就是那几个简单的示例,所以打算自己小结一下。
    一、io_service的作用 io_servie 实现了一个任务队列,这里的任务就是void(void)的函数。Io_servie最常用的两个接口是post和run,post向任务队列中投递任务,run是执行队列中的任务,直到全部执行完毕,并且run可以被N个线程调用。
    io_service类 你应该已经发现大部分使用Boost.Asio编写的代码都会使用几个io_service的实例。io_service是这个库里面最重要的类;它负责和操作系统打交道,等待所有异步操作的结束,然后为每一个异步操作调用其完成处理程序。
    boost boost::asio::read socket.read_some 区别
    boost boost::asio::read 尝试读一定数量的字节,直到读到为止,或者出错  socket.read_some 读一下socket,读到多少算多少  带async的类似