Boost Asio多线程应用定时器回调序列化问题排查

核心原因 #

你遇到的串行执行问题，本质是回调中的阻塞操作违反了Boost Asio的非阻塞设计原则。虽然你创建了2个工作线程，但Task1的回调里调用sleep(60)会直接阻塞当前工作线程，导致该线程无法处理其他任务。

如果实际运行中Task2必须等待Task1的sleep结束才执行，可能是以下场景之一：

• 单CPU核心环境下，系统线程调度策略导致Task2的线程被延迟调度；
• 标准输出缓冲导致你误判了输出顺序；
• 测试环境存在异常的线程调度限制。

但无论哪种情况，阻塞式回调都是不符合Asio设计理念的错误用法，会严重降低io_context的吞吐量。

改进建议 #

1. 用异步操作替代阻塞调用 #

将回调中的sleep(60)替换为Boost Asio的异步定时器，这样工作线程可以立即返回，继续处理其他任务：

timer.async_wait(
    [&ioContext](const boost::system::error_code& error) {
        if (error)
        {
            std::cerr << "async_wait failed: " << error.message() << std::endl;
            return;
        }
        std::cerr << "Sleep Start Task1 executed in thread: " << std::this_thread::get_id() << std::endl;
        // 用异步定时器替代阻塞sleep
        boost::asio::steady_timer delay_timer(ioContext, std::chrono::seconds(60));
        delay_timer.async_wait([](const boost::system::error_code& error) {
            if (!error) {
                std::cerr << "Task1 sleep finished in thread: " << std::this_thread::get_id() << std::endl;
            }
        });
    });

2. 用独立线程池处理阻塞任务 #

如果必须执行阻塞操作（比如调用第三方阻塞API），建议将这类任务提交到独立线程池，避免占用Asio的io_context工作线程：

// 在main中初始化独立线程池
std::vector<std::thread> blocking_pool;
for (int i = 0; i < 2; ++i) {
    blocking_pool.emplace_back([]() {
        // 线程池可扩展为带任务队列的实现，此处简化示例
        std::this_thread::sleep_for(std::chrono::hours(1));
    });
}

// 回调中提交阻塞任务
timer.async_wait(
    [](const boost::system::error_code& error) {
        if (error)
        {
            std::cerr << "async_wait failed: " << error.message() << std::endl;
            return;
        }
        std::cerr << "Sleep Start Task1 executed in thread: " << std::this_thread::get_id() << std::endl;
        // 提交阻塞任务到独立线程池
        std::async(std::launch::async, []() {
            sleep(60);
            std::cerr << "Task1 sleep finished in thread: " << std::this_thread::get_id() << std::endl;
        });
    });

3. 规范io_context使用细节 #

• 始终保证异步操作对象（如steady_timer）的生命周期长于回调，避免未定义行为；
• 如果回调涉及共享资源访问，使用io_context::strand序列化回调执行，保证线程安全。

修正后的完整代码示例 #

#include <boost/asio.hpp>
#include <iostream>
#include <vector>
#include <thread>
#include <chrono>
#include <future>

void worker(boost::asio::io_context& ioContext) {
    std::cerr << "Worker thread started." << std::endl;
    ioContext.run();
    std::cerr << "Worker thread stopped." << std::endl;
}

int main() {
    try {
        boost::asio::io_context ioContext;
        boost::asio::io_context::work work(ioContext);

        std::vector<std::thread> threads;
        for (int i = 0; i < 2; ++i) {
            threads.emplace_back(worker, std::ref(ioContext));
        }

        boost::asio::steady_timer timer(ioContext, std::chrono::seconds(5));
        timer.async_wait(
            [&ioContext](const boost::system::error_code& error) {
                if (error)
                {
                    std::cerr << "async_wait failed: " << error.message() << std::endl;
                    return;
                }
                std::cerr << "Sleep Start Task1 executed in thread: " << std::this_thread::get_id() << std::endl;
                // 异步延迟替代阻塞sleep
                boost::asio::steady_timer delay_timer(ioContext, std::chrono::seconds(60));
                delay_timer.async_wait([](const boost::system::error_code& error) {
                    if (!error) {
                        std::cerr << "Task1 sleep finished in thread: " << std::this_thread::get_id() << std::endl;
                    }
                });
            });

        boost::asio::steady_timer timer1(ioContext, std::chrono::seconds(10));
        timer1.async_wait(
            [](const boost::system::error_code& error) {
                if (error)
                {
                    std::cerr << "async_wait failed: " << error.message() << std::endl;
                    return;
                }
                std::cerr << "Start Task2 executed in thread: " << std::this_thread::get_id() << std::endl;
            });

        std::cerr << "App started :" << std::thread::hardware_concurrency() << std::endl;
        for (auto& thread : threads) {
            thread.join();
        }
        std::cerr << "All worker threads joined." << std::endl;
    } catch (const std::exception& e) {
        std::cerr << "Exception: " << e.what() << std::endl;
    }

    return 0;
}

内容的提问来源于stack exchange，提问作者root