个人笔记

外观

2026-11-09
1789
7.2m

Boost.Asio 多线程模型与组件封装：常驻 io_context、节流器与定时器 ​

用 Boost.Asio 搭一个带定时任务和节流回调的程序，绕不开两个基础问题：io_context 用什么方式组织到多个线程；定时器、节流器这类常用件怎么封装成可复用的组件。本文先把多线程组织的两种方式与各自的代价说清，再给出三个组件的完整封装：常驻后台线程的 io_context 单例、固定节拍的节流器、有限次数的周期定时器。

io_context 的两种多线程组织 ​

asio 程序从单线程扩展到多线程，有两种组织方式。

方式一：每个线程一个 io_context。通常让线程数与 CPU 核心数一致，每个线程跑自己 io_context 的事件循环。

  • 多核机器上能充分利用 CPU；
  • 同一个 socket 只会在一个 io_context 上出现，handler 天然被单线程串行执行，不需要额外同步；
  • handler 里不能执行阻塞操作，否则整个事件循环被卡住。

方式二：全局一个 io_context，多个线程共同 run()。异步事件完成时，handler 会被投递给任意一个正在 run() 的线程执行。

  • handler 里允许阻塞操作（例如数据库查询），此时可以把线程数开得比核心数大来掩盖阻塞；
  • 同一个 socket 的 handler 可能被不同线程并发执行，存在竞态：可读事件紧跟着发生两次时，两个线程可能同时读同一个 socket。解决办法是 strand——把需要串行化的 handler 挂到同一个 strand 上，asio 保证它们不并发执行。

选型上的一般规律：handler 都很轻、追求吞吐，用方式一避免同步开销；handler 里有阻塞调用或想简化线程管理，用方式二加 strand。

版本注记：旧代码里常见的 io_service 就是 io_context 的旧名——Boost 1.66 起被弃用并更名（io_service 保留为 deprecated typedef），Boost 1.87 起别名被移除，继续使用会编译失败。配套的 io_service::work 也由 executor_work_guard / make_work_guard 取代。新代码一律用 io_context。

常驻后台线程：io_context 单例 ​

程序里多个模块都需要 post 定时任务时，与其各建各的事件循环，不如共享一个 io_context，由单例统一持有：构造时起一个后台线程跑 run()，析构时 stop() 后 join()。

cpp
#include <boost/asio.hpp>
#include <thread>

class IoContextHolder
{
public:
    IoContextHolder()
        : work_guard_{boost::asio::make_work_guard(io_)}
    {
        thread_ = std::thread([this]() { io_.run(); });
    }

    ~IoContextHolder()
    {
        io_.stop();
        if (thread_.joinable())
            thread_.join();
    }

    static IoContextHolder& instance()
    {
        static IoContextHolder holder;  // Meyers 单例，C++11 起线程安全
        return holder;
    }

    boost::asio::io_context& context() { return io_; }

    // io_context 非拷贝非移动，单例也应禁拷贝
    IoContextHolder(const IoContextHolder&) = delete;
    IoContextHolder& operator=(const IoContextHolder&) = delete;

private:
    boost::asio::io_context io_;
    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> work_guard_;
    std::thread thread_;
};

两个关键点：

  1. make_work_guard 防止事件循环空转退出。io_context::run() 在没有待处理事件时会直接返回；work_guard 计数不为零时 run() 不会退出，即使暂时没有异步操作，后台线程也会一直待命。这也是它取代旧 io_service::work 的位置——旧写法是 std::unique_ptr<io_service::work> work(new io_service::work(io));，通过析构 work 让 run() 收尾。
  2. 关闭顺序：先 stop() 再 join()。stop 让 run() 尽快返回，此后挂起的 handler 不再被调用（这一行为是实测而非文档承诺，关闭顺序敏感的代码值得在自己的 Boost 版本上验证一次）。

各模块拿到 instance().context()，把自己的 timer、post 任务挂上去即可。

节流器：固定节拍，尾随最后一次调用 ​

高频事件（比如状态刷新、日志刷屏）希望最多每 interval 毫秒执行一次。节流器 Throttler 的语义是：

  • 距上次执行已满 interval：立即执行；
  • 未满 interval：安排在该满未满的时刻执行，期间的新调用会替换掉上一次推迟的计划——只有最后一次调用真正执行。

「替换」是靠 _timer.cancel() 实现的：cancel 会取消挂起的 async_wait，旧 handler 收到 operation_aborted 后不再执行回调；随后用新函数重排定时器。推迟时刻始终是「上次执行时间 + interval」，固定节拍，中间的调用全部合并到最后一次。

cpp
#include <boost/asio.hpp>
#include <chrono>
#include <functional>

class Throttler
{
public:
    using Clock = std::chrono::steady_clock;
    using Duration = std::chrono::milliseconds;

    Throttler(boost::asio::io_context& io, int interval_ms)
        : timer_(io), interval_(interval_ms)
    {
    }

    ~Throttler() { timer_.cancel(); }

    template <typename F, typename... Args>
    void execute(F&& f, Args&&... args)
    {
        auto now = Clock::now();
        auto remaining = interval_ - (now - previous_);  // 距下次可执行还有多久

        boost::system::error_code ec;
        timer_.cancel(ec);  // 取消上一次推迟的计划（若有）

        std::function<decltype(f(args...))()> func =
            std::bind(std::forward<F>(f), std::forward<Args>(args)...);

        if (remaining <= Duration::zero())
        {
            func();          // 节拍已过，立即执行
            previous_ = now;
        }
        else
        {
            timer_.expires_after(remaining);
            timer_.async_wait([this, func](const boost::system::error_code& ec) {
                if (!ec)     // 正常到期才执行；被 cancel 则跳过
                {
                    func();  // 回调在 IO 线程执行
                    previous_ = Clock::now();
                }
            });
        }
    }

private:
    boost::asio::steady_timer timer_;
    Duration interval_;
    Clock::time_point previous_{};  // 零值：首次调用必然立即执行
};

两个细节：

  • previous_ 值初始化为 clock 零点，首次调用时 remaining 为大负数，走立即执行分支——不需要特判首次；
  • 回调在 IO 线程执行，调用方与回调若访问同一状态，要么都经 io_context 串行化，要么自行加锁。

有限次数的周期定时器 ​

周期任务的标准写法是 handler 正常到期后重排 expires_after + async_wait，形成触发链。给它加上剩余次数计数，就是有限次数的定时器：

cpp
#include <boost/asio.hpp>
#include <chrono>
#include <functional>

class RepeatTimer
{
public:
    RepeatTimer(boost::asio::io_context& io, int timeout_ms, int repeat_times)
        : timer_(io), timeout_(timeout_ms), repeat_times_(repeat_times)
    {
    }

    ~RepeatTimer() { timer_.cancel(); }

    void start()
    {
        if (repeat_times_ == 0)
            return;
        timer_.expires_after(timeout_);
        timer_.async_wait([this](const boost::system::error_code& ec) { on_timeout(ec); });
    }

    template <typename F, typename... Args>
    void set_timeout_callback(F&& f, Args&&... args)
    {
        callback_ = std::bind(std::forward<F>(f), std::forward<Args>(args)...);
    }

private:
    void on_timeout(const boost::system::error_code& ec)
    {
        if (ec)
            return;  // 被 cancel，停止触发链

        --repeat_times_;
        if (repeat_times_ != 0)  // 次数未尽，先排下一轮
        {
            timer_.expires_after(timeout_);
            timer_.async_wait([this](const boost::system::error_code& e) { on_timeout(e); });
        }
        if (callback_)
            callback_();
    }

    boost::asio::steady_timer timer_;
    std::chrono::milliseconds timeout_;
    int repeat_times_;
    std::function<void()> callback_;
};

on_timeout 的顺序是先减计数、次数未尽先重排，再执行回调——重排放在回调前，回调耗时不占用周期。计数语义：构造 repeat_times = 3 则回调恰好触发 3 次，每次到期减一，减到零当次仍执行但不再重排。

这个组件与节流器共享同一个前提：每个组件持有自己的 steady_timer，但 io_context 由外部传入——组件不拥有事件循环，只挂在循环上。配合前面的 io_context 单例，程序里所有定时、节流、异步任务共用一条后台线程，互不干扰也不必各自管理线程生命周期。

结语 ​

  • io_context 两种组织：每线程一个，handler 免同步但不能阻塞；共享一个多线程 run()，handler 可阻塞但需 strand 串行化竞态；
  • io_service 旧名在 Boost 1.66 弃用、1.87 移除，io_service::work 的替代是 make_work_guard；
  • 常驻 io_context 单例 = work_guard 防退出 + 后台线程 run() + 析构 stop/join；
  • 节流器用 cancel + expires_after 实现「固定节拍、尾随最后一次调用」，timer 的 cancel 语义（旧 handler 收 operation_aborted）是替换计划的基础；
  • 周期定时器在 handler 里重排等待形成触发链，先重排后回调，计数控制次数。
最近更新

基于 VitePress 构建

Copyright © 2026 ww1820