用于deadline_timers的非阻塞提升io_service答案

【问题标题】：Non blocking boost io_service for deadline_timers用于deadline_timers的非阻塞提升io_service
【发布时间】：2014-11-03 18:07:56
【问题描述】：

阅读 boost::asio::deadline_timer 的文档后，似乎 io_service::run() 和处理程序方法是在同一个线程上调用的。在后台线程上运行io_service对象时，有什么方法可以在一个线程上创建计时器？

【问题讨论】：

你想达到什么目的？您可以根据需要使用服务运行任意数量的 IO 线程。所以，如果你想每隔一段时间给一个后台线程发信号，只需要引发一个条件变量或其他什么？
感谢您的回答。实际上，需要使进行回调的线程成为非阻塞线程。在正常实现中，一旦我们调用 io_service::run() 方法，它将被阻塞，直到时间计时器到期。所以两个线程都需要独立。

标签： c++ multithreading boost-asio

【解决方案1】：

为了有趣和荣耀，下面是如何将线程队列与 asio 截止时间计时器相结合，以从截止时间计时器分派非阻塞任务：

Live On Coliru

#ifndef HEADER_GUARD_CUSTOM_THREADPOOL_HPP
#define HEADER_GUARD_CUSTOM_THREADPOOL_HPP
#include <boost/function.hpp>
#include <boost/optional.hpp>
#include <boost/bind.hpp>
#include <boost/thread.hpp>
#include <boost/atomic.hpp>
#include <boost/phoenix.hpp>
#include <boost/make_shared.hpp>
#include <iostream>
#include <string>
#include <deque>

namespace custom {
    using namespace boost;

    class thread_pool
    {
    private:
        mutex mx;
        condition_variable cv;

        typedef function<void()> job_t;
        std::deque<job_t> _queue;

        thread_group pool;

        boost::atomic_bool shutdown;
        static void worker_thread(thread_pool& q)
        {
            while (optional<job_t> job = q.dequeue())
                (*job)();
        }

    public:
        thread_pool() : shutdown(false) {
            //LOG_INFO_MESSAGE << "Number of possible Threads: " << boost::thread::hardware_concurrency() << std::endl;
            for (unsigned i = 0; i < boost::thread::hardware_concurrency(); ++i){
                pool.create_thread(bind(worker_thread, ref(*this)));
            }
        }

        void enqueue(job_t job)
        {
            lock_guard<mutex> lk(mx);
            _queue.push_back(job);

            cv.notify_one();
        }

        optional<job_t> dequeue()
        {
            unique_lock<mutex> lk(mx);
            namespace phx = boost::phoenix;

            cv.wait(lk, phx::ref(shutdown) || !phx::empty(phx::ref(_queue)));

            if (_queue.empty())
                return none;

            job_t job = _queue.front();
            _queue.pop_front();
            return job;
        }

        ~thread_pool()
        {
            shutdown = true;
            {
                lock_guard<mutex> lk(mx);
                cv.notify_all();
            }

            pool.join_all();
        }
    };
}

#endif // HEADER_GUARD_CUSTOM_THREADPOOL_HPP

还有简单的测试程序：

#include <boost/asio.hpp>

namespace a = boost::asio;
using error = boost::system::error_code;

void timer_loop(a::deadline_timer& tim, custom::thread_pool& pool) {
    static boost::atomic_int count(0);

    tim.expires_from_now(boost::posix_time::milliseconds(10));
    tim.async_wait([&](error ec) {
        if (!ec && (++count < 100)) {
            int id = count;

            pool.enqueue([id] { 
                std::cout << "timer callback " << id << " started on thread " << boost::this_thread::get_id() << "\n";
                boost::this_thread::sleep_for(boost::chrono::milliseconds(rand()%1000));
                std::cout << "timer callback " << id << " completed\n";
            });

            std::cout << "Job " << id << " enqueued" << "\n";
            timer_loop(tim, pool);
        }
    });
}

int main()
{
    a::io_service svc;
    a::deadline_timer tim(svc);
    custom::thread_pool pool;

    timer_loop(tim, pool);

    svc.run();
}

【讨论】：