[C++知识库]学习 浅谈 Boost.Asio 的多线程模型的

最近想了解一下多线程和线程池，就简单的看了一下不是很精，看到啥就写啥吧

首先贴一下看的文章http://senlinzhan.github.io/2017/09/17/boost-asio/

github的代码仓库https://github.com/senlinzhan/code-for-blog

贴一下我写的简单注解

#include "asio_io_service_pool.hpp"

#include <boost/asio.hpp>
#include <boost/asio/steady_timer.hpp>
#include <chrono>
#include <iostream>
#include <mutex>

int main()
{
    AsioIOServicePool pool(4);
    // 多个线程拥有自己的IO， run自己的任务
    boost::asio::steady_timer timer1{pool.getIOService(), std::chrono::seconds{2}};
    boost::asio::steady_timer timer2{pool.getIOService(), std::chrono::seconds{2}};

    int value = 0;
    std::mutex mtx;    // 互斥锁，用来保护 std::cout 和 value

    timer1.async_wait([&mtx, &value] (const boost::system::error_code &ec)
                      {
                          std::lock_guard<std::mutex> lock(mtx);
                          std::cout << "Hello, World! " << value++ << std::endl;
                      });

    timer2.async_wait([&mtx, &value] (const boost::system::error_code &ec)
                      {
                          std::lock_guard<std::mutex> lock(mtx);
                          std::cout << "Hello, World! " << value++ << std::endl;
                      });
    pool.stop();

    return 0;
}

#ifndef ASIO_IO_SERVICE_POOL_H
#define ASIO_IO_SERVICE_POOL_H

#include <boost/asio.hpp>
#include <memory>
#include <thread>
#include <vector>

//在多核的机器上，这种方案可以充分利用多个 CPU 核心。
//某个 socket 描述符并不会在多个线程之间共享，每个线程都有各自的Io_Servivce,有各自的socket，所以不需要引入同步机制。
//在 event handler 中不能执行阻塞的操作，否则将会阻塞掉io_service所在的线程。
// http://senlinzhan.github.io/2017/09/17/boost-asio/
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;

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

#endif /* ASIO_IO_SERVICE_POOL_H */

#include "asio_thread_pool.hpp"

#include <boost/asio.hpp>
#include <boost/asio/steady_timer.hpp>
#include <chrono>
#include <iostream>

int main()
{
    // 通过strand.wrap将callback函数进行串行化，即时是多个线程也保证了顺序，相当于单线程
    AsioThreadPool pool(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();

    return 0;
}


/*Asio 多线程无锁串行化
单 io_service 多线程模式时 io_service 的典型用法，在这种模式下，多个线程会竞争 io_service，竞争到的线程会得到处理下一个 handler 的机会，通过这种用法，表现为 io_service 会自动的将 io 事件分配给各个线程去处理，并且我们自己的任务也可以通过 io_service.post() 丢给线程们去执行，这相当于我们拥有了一个线程池。

io_service 本身是线程安全的，多个线程在竞争 io_service 时会加锁，然而这只是在竞争 io_service 时加锁，在任何一个线程拿到 io_service 并开始执行 handler 后，io_service 的锁会释放交给其他线程继续竞争。所以可以看到，handlers 是多线程并发执行的，在各个 handlers 中如果存在共享的数据，我们仍是需要处理同步问题。

为此 io_service 提供了 strand，strand 可以用来包裹 handlers，被同一个 strand 包裹的 handlers 会串行执行，这样就可以省的我们自己在 handlers 为共享数据加锁了。实际上，当我们只在一个线程上执行 io_service.run() 时，所有的 handlers 也是通过一个 strand 串行执行的，这在 asio 文档中被称为 "implicit strand".

可以想象，如果我们在多线程单 io_service 的用法下，对所有 handlers 使用同一个 strand 包裹，那么这些 handlers 也将全部串行执行，这将和单线程单 io_service 没有区别，多线程在这种情况下就没有意义了。

另外要注意 strand 是属于某个 io_service 的，也就是说 strand 只能串行化它属于的 io_service 下的 handlers。
*/

#include "asio_thread_pool.hpp"

#include <boost/asio.hpp>
#include <boost/asio/steady_timer.hpp>
#include <chrono>
#include <iostream>
#include <mutex>

int main()
{
    AsioThreadPool pool(4);    // 开启 4 个线程
    // 同一个IO service 在多个线程中run， event_handle 在多个线程中回调
    // 比如socket 再两个线程中同时读操作，那就问题了 可以用strand.wrap()解决

    boost::asio::steady_timer timer1{pool.getIOService(), std::chrono::seconds{2}};
    boost::asio::steady_timer timer2{pool.getIOService(), std::chrono::seconds{2}};

    int value = 0;
    std::mutex mtx;    // 互斥锁，用来保护 std::cout 和 value

    timer1.async_wait([&mtx, &value] (const boost::system::error_code &ec)
                      {
                          std::lock_guard<std::mutex> lock(mtx);
                          std::cout << "Hello, World! " << value++ << std::endl;
                      });

    timer2.async_wait([&mtx, &value] (const boost::system::error_code &ec)
                      {
                          std::lock_guard<std::mutex> lock(mtx);
                          std::cout << "Hello, World! " << value++ << std::endl;
                      });
    pool.stop();

    return 0;
}

#ifndef THREAD_POOL_H
#define THREAD_POOL_H

#include <boost/asio.hpp>
#include <memory>
#include <thread>
#include <vector>

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

#endif /* THREAD_POOL_H */

看到里面的CmakeList， 想自己要不编一下，于是就开始弄起cmake了， 建了个build目录，cmake .. 产生makefile，失败找不到boost库，然后安装 apt-get install libboost-all-dev, 删掉几个没有的文件夹，就开始make了，修改编译错误就ok了

这里回顾了一下 Cmake 用来产生makefile 文件并不是编译， 当make的时候才是编译的开始

这里有一个疑问没搞明白就是，我CmakeList target_link_libraries ${Boost_SYSTEM_LIBRARY},为什么不需要include_directions()就可以用boost库了，他自己怎么找到同文件目录的 不明白