Bost asio库与线程池的使用

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

每个线程一个 I/O Service

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

在多核的机器上，这种方案可以充分利用多个 CPU 核心。 某个 socket 描述符并不会在多个线程之间共享，所以不需要引入同步机制。 在 event handler 中不能执行阻塞的操作，否则将会阻塞掉io_service所在的线程。 直接上代码： 

讯享网#include <boost/asio.hpp> #include <string> #include<memory> #include <boost/unordered_map.hpp> class ThreadPool { 
    public: typedef boost::asio::io_context ioContext; typedef boost::asio::io_context::work Work; ThreadPool(std::size_t size = std::thread::hardware_concurrency()) :my_pool_size{ 
   size} { 
    for (std::size_t i = 0; i < size; ++i){ 
    std::shared_ptr<ioContext> io_context = std::make_shared<ioContext>(); std::shared_ptr<Work> work =std::make_shared<Work>(); m_ioContextList.push_back(io_context); m_workList.push_back(work); } //开启线程池,一个线程一个I/O 服务 for (int i = 0; i < size ; i++) { 
    std::thread runThread([=] { 
    m_ioContextList[i]->run(); }); runThread.detach(); //runThread.post(); } } // 使用 round-robin 的方式返回一个 io_service boost::asio::io_context &getIoContext() { 
    //通过这个函数分配io_context  ioContext& io_context = *m_ioContextList[m_io_context_pos++]; if (m_io_context_pos == my_pool_size - 1) m_io_context_pos = 0; return io_context; } void stop(){ 
    for (int i = 0; i < my_pool_size; i++) { 
    m_ioContextList[i]->stop(); } } private: std::vector<std::shared_ptr<ioContext>> m_ioContextList; std::vector<std::shared_ptr<Work>> m_workList; size_t my_pool_size; size_t m_io_context_pos; //依次分配io_context }; 

一个 I/O Service 与多个线程

讯享网

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

也是直接上代码：

讯享网#include <functional> #include <iostream> #include <mutex> #include <thread> #include <vector> #define BOOST_ASIO_NO_DEPRECATED #include <boost/thread/thread.hpp> #include <boost/asio.hpp> class ThreadPool { 
    public: // the constructor just launches some amount of threads explicit ThreadPool(std::size_t size) : io_context_(size), strand_(io_context_), work_guard_(boost::asio::make_work_guard(io_context_)) { 
    // one io_context and multi-thread for (std::size_t i = 0; i < size; ++i) { 
    // all the threads do is execute the io_context::run() group_.create_thread([&](){ 
    io_context_.run(); }); } } // the destructor joins all threads ~ThreadPool() { 
    // Once the work object is destroyed, the service will stop. work_guard_.reset(); group_.join_all(); } // Add new work item to the pool. template<class F> void Enqueue(F f) { 
    // Submits a completion token or function object for execution. boost::asio::post(io_context_, f); } private: boost::thread_group group_; boost::asio::io_context io_context_; boost::asio::io_context::strand strand_; // prevent the run() method from return. typedef boost::asio::io_context::executor_type ExecutorType; boost::asio::executor_work_guard<ExecutorType> work_guard_; }; // For output. std::mutex g_io_mutex; int main ( int argc, char* argv[] ) { 
    int thread_num = std::thread::hardware_concurrency(); std::cout << "thread num: " << thread_num<< std::endl; ThreadPool pool(thread_num); // Queue a bunch of work items. for (int i = 0; i < 4; ++i) { 
    pool.Enqueue([i] { 
    { 
    std::lock_guard<std::mutex> lock(g_io_mutex); std::cout << "Hello" << "(" << i << ") " << std::endl; } std::this_thread::sleep_for(std::chrono::seconds(1)); { 
    std::lock_guard<std::mutex> lock(g_io_mutex); std::cout << "World" << "(" << i << ")" << std::endl; } }); } return 0; } 

运行结果

参考:http://senlinzhan.github.io/2017/09/17/boost-asio/