深入学习 Boost.Asio（一）：从原理到 io_context

1
2
3
4

线程1: read(socket_1) ← 阻塞等待用户1输入...
线程2: read(socket_2) ← 阻塞等待用户2输入...
...
线程1000: read(socket_1000) ← 阻塞等待用户1000输入...

1
2
3
4
5
6

单线程事件循环:
  ┌→ 等待事件（epoll/IOCP）
  │   ├─ socket_7 可读 → 处理用户7的消息
  │   ├─ socket_42 可读 → 处理用户42的消息
  │   └─ socket_100 可写 → 继续发送给用户100
  └─ 回到等待

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

Reactor（反应器）模式：
  应用                    OS
   │                      │
   │ "告诉我 socket      │
   │  什么时候可读"       │
   ├─────────────────────→│
   │                      │ ...等待...
   │  "socket_7 可读了"   │
   │←─────────────────────┤
   │                      │
   │ read(socket_7, buf)  │  ← 应用自己执行读操作
   ├─────────────────────→│
   │  返回数据            │
   │←─────────────────────┤

Proactor（前摄器）模式：
  应用                    OS
   │                      │
   │ "帮我读 socket，     │
   │  读完了通知我"       │
   ├─────────────────────→│
   │                      │ ...OS 执行读取...
   │  "读完了，数据在     │
   │   这个 buffer 里"    │
   │←─────────────────────┤

1
2
3
4
5
6

// Proactor 模式下，用户代码更简洁：
// 你只需要关心"操作完成后做什么"，不用关心中间的 I/O 细节
socket.async_read_some(buffer,
    [](error_code ec, size_t bytes_transferred) {
        // 到这里时，数据已经在 buffer 里了
    });

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// Reactor 模式下，你需要自己处理"就绪"到"完成"之间的逻辑
reactor.register_read(socket, [&]() {
    // socket 可读了，但你需要自己读
    ssize_t n = ::read(socket.native_handle(), buf, sizeof(buf));
    if (n == -1 && errno == EAGAIN) {
        // 哎呀，虽然说可读，但实际上没数据（虚假唤醒）
        return;
    }
    // 处理数据...
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

┌─────────────────────────────────────────────────┐
│                  io_context                       │
│                                                  │
│  ┌────────────────────────┐                      │
│  │  scheduler (调度器)     │                      │
│  │  ├─ op_queue (操作队列) │ ← 就绪的完成回调    │
│  │  ├─ timer_queue         │ ← 定时器堆          │
│  │  └─ reactor             │ ← epoll/IOCP 封装   │
│  └────────────────────────┘                      │
│                                                  │
│  ┌────────────────────────┐                      │
│  │  work_count (原子计数)  │ ← 决定 run() 是否退出│
│  └────────────────────────┘                      │
└─────────────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// io_context::run() 的简化伪代码
void io_context::run() {
    while (true) {
        // 1. 如果 work_count == 0 且操作队列为空 → 退出
        if (stopped_ || (work_count_ == 0 && op_queue_.empty()))
            break;

        // 2. 从操作队列取一个就绪的完成处理器执行
        if (auto op = op_queue_.dequeue()) {
            op->complete();  // 执行用户的回调/恢复协程
            continue;
        }

        // 3. 操作队列为空，阻塞在 reactor 上等待新事件
        //    （epoll_wait / GetQueuedCompletionStatus）
        reactor_.poll(op_queue_);
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

// 示例：观察 io_context 的任务调度行为
#include <boost/asio.hpp>
#include <iostream>
#include <thread>

int main()
{
    boost::asio::io_context ioCtx;

    // 向 io_context 投递两个任务
    boost::asio::post(ioCtx, []() {
        std::cout << "[任务1] 线程ID: "
                  << std::this_thread::get_id() << "\n";
    });

    boost::asio::post(ioCtx, []() {
        std::cout << "[任务2] 线程ID: "
                  << std::this_thread::get_id() << "\n";
    });

    boost::asio::post(ioCtx, []() {
        std::cout << "[任务3] 线程ID: "
                  << std::this_thread::get_id() << "\n";
    });

    // 此时队列中有 3 个任务
    // run() 会逐个取出执行，全部完成后返回
    std::cout << "=== run() 开始 ===\n";
    ioCtx.run();
    std::cout << "=== run() 结束 ===\n";

    // 输出：
    // === run() 开始 ===
    // [任务1] 线程ID: 1
    // [任务2] 线程ID: 1
    // [任务3] 线程ID: 1
    // === run() 结束 ===
    //
    // 观察：三个任务在同一个线程中顺序执行

    return 0;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

boost::asio::io_context ioCtx;

// post 几个任务
boost::asio::post(ioCtx, []() { std::cout << "A\n"; });
boost::asio::post(ioCtx, []() { std::cout << "B\n"; });
boost::asio::post(ioCtx, []() { std::cout << "C\n"; });

// run_one()：执行一个任务就返回（如果没有就绪任务则阻塞等待）
ioCtx.run_one();  // 输出 "A"

// poll()：执行所有当前就绪的任务，不阻塞（没有任务就立即返回）
ioCtx.poll();     // 输出 "B" 和 "C"

// poll_one()：执行一个就绪任务，不阻塞
ioCtx.poll_one(); // 无任务，立即返回（返回值 0）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// 游戏服务器中典型的 poll() 用法
void gameLoop()
{
    boost::asio::io_context ioCtx;
    // ... 设置网络监听 ...

    while (running)
    {
        // 处理所有就绪的网络事件（不阻塞）
        ioCtx.poll();

        // 更新游戏逻辑（每帧固定执行）
        updateGameWorld();

        // 控制帧率
        sleepUntilNextFrame();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

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

int main()
{
    boost::asio::io_context ioCtx;

    // 场景1：没有 work_guard
    // ioCtx.run();  // ← 立即返回！因为没有任何任务

    // 场景2：有 work_guard
    // work_guard 内部将 work_count 加 1，使得 run() 认为"还有工作"
    auto guard = boost::asio::make_work_guard(ioCtx);

    // 在另一个线程中模拟"5秒后才开始有连接进来"
    std::thread timer([&]() {
        std::this_thread::sleep_for(std::chrono::seconds(5));
        // 假设此时开始有工作了，释放 guard
        guard.reset();  // work_count 减 1，允许 run() 退出
    });

    std::cout << "服务器启动，等待连接...\n";
    ioCtx.run();  // 阻塞 5 秒后退出
    std::cout << "事件循环退出\n";

    timer.join();
    return 0;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

用户代码                    Asio 内部                      操作系统
   │                          │                              │
   │ async_read_some(buf, h)  │                              │
   ├─────────────────────────→│                              │
   │                          │                              │
   │                          │ 1. 创建 read_op 对象         │
   │                          │    (包含 buffer 指针和       │
   │                          │     handler 的拷贝)          │
   │                          │                              │
   │                          │ 2. 尝试立即完成              │
   │                          │    (非阻塞 recv)             │
   │                          ├─────────────────────────────→│
   │                          │                              │
   │                    ┌─────┤ 3a. 如果有数据 → 直接入队    │
   │                    │     │     handler 到 op_queue      │
   │                    │     │                              │
   │                    └─or──┤ 3b. 如果 EAGAIN → 注册到     │
   │                          │     reactor (epoll_ctl)      │
   │                          ├─────────────────────────────→│
   │                          │                              │
   │ (async_read_some 立即返回│                              │
   │  用户代码继续执行)        │                              │
   │                          │                              │
   │                          │ ... 时间流逝 ...             │
   │                          │                              │
   │                          │ 4. reactor 检测到可读         │
   │                          │←─────────────────────────────┤
   │                          │                              │
   │                          │ 5. 执行实际 recv()           │
   │                          ├─────────────────────────────→│
   │                          │    数据写入 buffer           │
   │                          │←─────────────────────────────┤
   │                          │                              │
   │                          │ 6. 将完成的 handler 入队     │
   │                          │    op_queue                  │
   │                          │                              │
   │ 7. run() 取出 handler 执行│                              │
   │←─────────────────────────┤                              │
   │                          │                              │
   │ handler(ec, bytes_read)  │                              │

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// 演示：async 操作立即返回，不会阻塞调用线程
#include <boost/asio.hpp>
#include <iostream>

using boost::asio::ip::tcp;

int main()
{
    boost::asio::io_context ioCtx;

    tcp::acceptor acceptor(ioCtx, tcp::endpoint(tcp::v4(), 9000));

    std::cout << "1. 发起 async_accept...\n";

    // async_accept 立即返回！不会等待连接到来
    acceptor.async_accept([](boost::system::error_code ec, tcp::socket socket) {
        std::cout << "4. 有连接到来！\n";
    });

    std::cout << "2. async_accept 已返回（此时还没有连接）\n";
    std::cout << "3. 开始 run()，在这里阻塞等待事件...\n";

    // run() 内部会阻塞在 epoll_wait/IOCP 上
    // 直到有客户端连接，handler 被调用后，没有更多任务，run() 退出
    ioCtx.run();

    std::cout << "5. run() 退出\n";
    return 0;
}
// 输出顺序：1 → 2 → 3 → (等待连接) → 4 → 5

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// ✅ 好的 handler：轻量、快速返回
socket.async_read_some(buffer,
    [](boost::system::error_code ec, size_t n) {
        // 快速处理数据，然后发起下一个异步操作
        // 不要在这里做耗时计算！
    });

// ❌ 糟糕的 handler：阻塞事件循环
socket.async_read_some(buffer,
    [](boost::system::error_code ec, size_t n) {
        // 在 handler 里做耗时操作会阻塞整个事件循环！
        std::this_thread::sleep_for(std::chrono::seconds(5));
        // 这 5 秒内，所有其他连接都无法处理
    });

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

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

int main()
{
    boost::asio::io_context ioCtx;

    // 创建一个 2 秒后到期的定时器
    boost::asio::steady_timer timer(ioCtx, std::chrono::seconds(2));

    std::cout << "定时器已设置，等待 2 秒...\n";

    // 异步等待：timer 到期时执行 lambda
    timer.async_wait([](boost::system::error_code ec) {
        if (!ec) {
            std::cout << "定时器到期！\n";
        } else if (ec == boost::asio::error::operation_aborted) {
            // 定时器被 cancel() 或析构前被取消
            std::cout << "定时器被取消\n";
        }
    });

    ioCtx.run();
    return 0;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

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

// 用类封装周期定时器，避免生命周期问题
class PeriodicTimer
{
public:
    // 构造函数：指定 io_context、间隔时间、最大执行次数
    PeriodicTimer(boost::asio::io_context& ioCtx,
                  std::chrono::milliseconds interval,
                  int maxCount)
        : timer_(ioCtx)       // 定时器绑定到 io_context
        , interval_(interval) // 保存间隔
        , maxCount_(maxCount) // 最大次数
        , count_(0)           // 当前计数
    {
    }

    // 启动定时器
    void start()
    {
        // 设置到期时间为"当前时间 + 间隔"
        timer_.expires_after(interval_);

        // 异步等待，到期后执行 onTimer
        timer_.async_wait([this](boost::system::error_code ec) {
            onTimer(ec);
        });
    }

private:
    void onTimer(boost::system::error_code ec)
    {
        if (ec) return;  // 被取消或出错，不再继续

        ++count_;
        std::cout << "第 " << count_ << " 次触发\n";

        if (count_ < maxCount_)
        {
            // 关键：重新设置到期时间，再次等待
            // 使用 expires_at 而非 expires_after 可以避免时间漂移
            timer_.expires_at(timer_.expiry() + interval_);
            timer_.async_wait([this](boost::system::error_code ec) {
                onTimer(ec);
            });
        }
        // 如果 count_ >= maxCount_，不再设置新的等待
        // io_context 没有其他任务时 run() 会退出
    }

    boost::asio::steady_timer timer_;
    std::chrono::milliseconds interval_;
    int maxCount_;
    int count_;
};

int main()
{
    boost::asio::io_context ioCtx;

    // 每 500ms 触发一次，共 5 次
    PeriodicTimer pt(ioCtx, std::chrono::milliseconds(500), 5);
    pt.start();

    ioCtx.run();  // 2.5 秒后退出
    std::cout << "完成\n";
    return 0;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

#include <boost/asio.hpp>
#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <iostream>
#include <chrono>

using boost::asio::awaitable;
using boost::asio::use_awaitable;

// 协程式周期定时器：代码像同步循环一样直观
awaitable<void> periodicTask(int intervalMs, int count)
{
    // 从协程的执行器创建定时器
    auto executor = co_await boost::asio::this_coro::executor;
    boost::asio::steady_timer timer(executor);

    for (int i = 1; i <= count; ++i)
    {
        // 设置到期时间
        timer.expires_after(std::chrono::milliseconds(intervalMs));

        // co_await 挂起当前协程，定时器到期后自动恢复
        co_await timer.async_wait(use_awaitable);

        std::cout << "第 " << i << " 次触发\n";
    }
    std::cout << "周期任务结束\n";
}

int main()
{
    boost::asio::io_context ioCtx;
    boost::asio::co_spawn(ioCtx, periodicTask(500, 5), boost::asio::detached);
    ioCtx.run();
    return 0;
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

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

int main()
{
    boost::asio::io_context ioCtx;
    boost::asio::steady_timer timer(ioCtx, std::chrono::seconds(10));

    timer.async_wait([](boost::system::error_code ec) {
        if (ec == boost::asio::error::operation_aborted) {
            std::cout << "定时器被取消了！\n";
        } else {
            std::cout << "定时器正常到期\n";
        }
    });

    // 取消定时器：所有等待中的 handler 会以 operation_aborted 被调用
    std::size_t cancelled = timer.cancel();
    std::cout << "取消了 " << cancelled << " 个等待中的操作\n";

    ioCtx.run();
    return 0;
}
// 输出：
// 取消了 1 个等待中的操作
// 定时器被取消了！

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

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

int main()
{
    boost::asio::io_context ioCtx;
    auto guard = boost::asio::make_work_guard(ioCtx);

    // 在另一个线程运行 io_context
    std::thread worker([&ioCtx]() { ioCtx.run(); });

    // === post：总是排入队列 ===
    // 无论当前线程是不是 io_context 线程，都排队等 run() 执行
    boost::asio::post(ioCtx, []() {
        std::cout << "[post] 一定在 io_context 线程中执行\n";
    });

    // === dispatch：看情况 ===
    // 如果当前线程正在 run() io_context → 立即执行
    // 否则 → 等同于 post
    boost::asio::dispatch(ioCtx, []() {
        std::cout << "[dispatch] 可能立即执行，也可能排队\n";
    });

    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    guard.reset();
    worker.join();
    return 0;
}

1
2
3
4
5
6
7
8

调用 dispatch(fn)
      │
当前线程正在运行此 io_context 的 run()？
 ┌────┴────┐
是         否
 │          │
立即执行   等同于 post
fn()       （排入队列）

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

class Connection
{
public:
    // 这个函数可能从任意线程调用
    void send(const std::string& msg)
    {
        // dispatch 保证：
        // - 如果已经在 io_context 线程 → 直接发送（零开销）
        // - 如果在其他线程 → 安全地投递到 io_context 线程
        boost::asio::dispatch(ioCtx_, [this, msg]() {
            doSend(msg);  // 一定在 io_context 线程中执行
        });
    }

private:
    void doSend(const std::string& msg) { /* 实际发送逻辑 */ }
    boost::asio::io_context& ioCtx_;
};