跳到正文

Wiki

Boost.Asio 基礎

約 9 分鐘閱讀

本文由簡體中文內容確定性轉換,並受版本化術語表保護。

本節不急着講串口、TCP、UDP,而是把 io_context、run()、post()、work_guard、buffer、std::bind、對象生命週期這些基礎講清楚。
這些東西如果沒搞懂,後面串口和網絡程序會出現“為什麼回調不執行”“為什麼程序直接退出”“為什麼段錯誤”等問題。


1.示例 1:io_context 沒有任務時,run() 立刻返回

1.1.程序目標

驗證一個非常重要的現象:如果 io_context 裏沒有任何未完成任務,io.run() 會立刻返回。

1.2.完整代碼

#include <boost/asio.hpp>
#include <print>

int main()
{
    // 程序从 main 函数开始执行,下面的语句会按顺序运行。
    // io_context 是 Asio 的事件循环对象,异步任务需要靠它调度。
    boost::asio::io_context io;

    std::println("main:准备调用 io.run()");

    // 启动事件循环,前面注册的异步任务会在这里被调度执行。
    std::size_t count = io.run();

    std::println("main:io.run() 返回,执行了 {} 个回调", count);

    return 0;
}

運行結果:見下方“運行輸出與時間順序”;如果示例涉及定時器、線程、網絡或外部設備,具體時間和順序可能會隨環境略有變化。

1.3.運行輸出與時間順序

這個程序不會等待,立刻輸出:

main:准备调用 io.run()
main:io.run() 返回,执行了 0 个回调

1.4.本示例需要注意的點

io.run() 不是永遠阻塞。它阻塞的前提是:

io_context 里还有未完成的异步任务,或者还有 work_guard 保持它不退出。

如果什麼任務都沒有,它會馬上返回。

1.5.io.run() 返回值

io.run() 返回值表示執行了多少個 handler。

本例中沒有任何任務,所以返回值是:

0

2.示例 2:普通 main() 裏用 post() 投遞任務

2.1.程序目標

boost::asio::post() 可以把一個普通函數投遞到 io_context 裏,讓它以後由 io.run() 執行。

2.2.完整代碼

#include <boost/asio.hpp>
#include <functional>
#include <print>
#include <string>

void print_msg(const std::string& msg)
{
    std::println("执行任务:{}", msg);
}

int main()
{
    // 程序从 main 函数开始执行,下面的语句会按顺序运行。
    // io_context 是 Asio 的事件循环对象,异步任务需要靠它调度。
    boost::asio::io_context io;

    std::println("main:投递任务 A");
    // post 只是把任务放进队列,真正执行要等 io.run()。
    boost::asio::post(io, std::bind(print_msg, std::string("A")));

    std::println("main:投递任务 B");
    boost::asio::post(io, std::bind(print_msg, std::string("B")));

    std::println("main:准备调用 io.run()");

    // 启动事件循环,前面注册的异步任务会在这里被调度执行。
    std::size_t count = io.run();

    std::println("main:io.run() 返回,执行了 {} 个任务", count);

    return 0;
}

運行結果:見下方“運行輸出與時間順序”;如果示例涉及定時器、線程、網絡或外部設備,具體時間和順序可能會隨環境略有變化。

2.3.運行輸出與時間順序

程序立刻輸出:

main:投递任务 A
main:投递任务 B
main:准备调用 io.run()
执行任务:A
执行任务:B
main:io.run() 返回,执行了 2 个任务

2.4.本示例需要注意的點

post() 不是立刻執行函數,而是把函數放進 io_context 的任務隊列。

真正執行發生在:

io.run();

2.5.post() 函數説明

boost::asio::post(io, handler);

參數:

  1. io:任務投遞到哪個 io_context;
  2. handler:將來要執行的函數對象。

返回值:通常不用關心。

2.6.std::bind 説明

std::bind(print_msg, std::string("A"))

表示生成一個“無參數函數對象”。以後執行它時,相當於執行:

print_msg("A");

3.示例 3:work_guard 讓 io.run() 等待未來任務

3.1.程序目標

首先要理解 io.run() 一個非常重要的規則:

io.run() 不是一個永遠運行的死循環。

當 io_context 發現:

现在没有 handler 可以执行

并且

以后也没有已经登记的异步工作需要等待

那麼:

io.run();

就會直接返回。

例如:

boost::asio::io_context io;

io.run();

這裏什麼任務都沒有,所以:

进入 run()
↓
发现没有任何工作
↓
立即返回

這對於機械人通信線程就可能產生一個問題。

假設程序啓動時先創建 Asio 線程:

通信线程
↓
io.run()

但是串口、定時器或者其他任務要等一會兒才從別的線程提交。

那麼可能發生:

Asio 线程启动
↓
io.run()
↓
此时任务队列为空
↓
run() 认为已经没事可做
↓
直接返回
↓
Asio 通信线程结束

过了 2 秒
↓
其他线程才 post() 一个任务
↓
已经没有线程在 run() 了
↓
这个任务没人执行

work_guard 就是為了解決這種情況。

可以先把它理解成:

給 io_context 人為登記一份“還有工作沒有完成”的佔位工作。

只要這個 guard 還有效:

io_context:
虽然现在任务队列是空的,

但还有一份 outstanding work,

所以我不能认为工作已经全部结束。

於是:

io.run();

不會因為“暫時沒任務”而返回,而是繼續阻塞等待以後可能到來的任務。


3.2.如果沒有 work_guard

先看這個過程:

boost::asio::io_context io;

io.run();

此時:

任务数量 = 0
未完成异步操作 = 0
work_guard = 0

所以:

io.run()
↓
没有任何工作
↓
直接返回

可以把 run() 粗略理解成:

while (还有工作需要处理)
{
    等待事件
    执行 handler
}

return;

注意這裏的:

还有工作

不只是指:

当前任务队列里已经有 handler

也包括:

已经启动但还没有完成的异步操作

例如:

timer.async_wait(...);

雖然 timer 還要 2 秒才到期,但 Asio 已經知道:

这里还有一个异步操作没完成

所以這種情況下:

io.run();

本來就不會立刻返回。

這也是為什麼用 timer.async_wait() 來演示 work_guard 並不好:

timer 自己已經能夠讓 io.run() 保持運行,反而看不出 guard 的作用。


3.3.work_guard 到底做了什麼

創建:

auto guard = boost::asio::make_work_guard(io);

以後,可以粗略理解成:

io_context 内部的“未完成工作数量”
额外 +1

注意,這裏的 +1 只是幫助理解,並不代表內部真的就是一個簡單整數。

可以想像成:

创建 io_context

未完成工作:
0

創建 guard:

make_work_guard(io)

未完成工作:
0 + 一份 guard 占位工作

於是即使現在:

没有 timer

没有 socket

没有 post 的 handler

没有任何真正任务

io.run() 看到 guard 還持有 outstanding work,也不會退出。

它會:

没有任务
↓
但是 guard 还在
↓
不能返回
↓
阻塞等待新任务

注意:

work_guard 不會讓 io.run() 瘋狂空轉。

它不是:

while (true)
{
}

沒有任務時,io.run() 仍然可以阻塞等待,不會因為 guard 就白白佔滿一個 CPU 核心。


3.4.完整代碼

這個例子真正模擬:

Asio 通信线程已经启动

但当前没有任何任务

2 秒以后另一个线程才提交任务
#include <boost/asio.hpp>
#include <chrono>
#include <functional>
#include <print>
#include <thread>

using namespace std::chrono_literals;

using WorkGuard =
    boost::asio::executor_work_guard<
        boost::asio::io_context::executor_type>;

void delayed_post(
    boost::asio::io_context* io,
    WorkGuard* guard)
{
    std::println("工作线程:等待 2 秒再提交任务");

    std::this_thread::sleep_for(2s);

    std::println("工作线程:向 io_context 提交任务");

    boost::asio::post(
        *io,
        [] {
            std::println(
                "Asio handler:收到后来提交的任务");
        });

    std::println("工作线程:释放 work_guard");

    guard->reset();
}

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

    WorkGuard guard =
        boost::asio::make_work_guard(io);

    std::thread worker(
        std::bind(
            delayed_post,
            &io,
            &guard));

    std::println("main:开始 io.run()");

    std::size_t count = io.run();

    std::println(
        "main:io.run() 返回,执行了 {} 个 handler",
        count);

    worker.join();

    return 0;
}

運行結果大致為:

main:开始 io.run()
工作线程:等待 2 秒再提交任务

工作线程:向 io_context 提交任务
工作线程:释放 work_guard
Asio handler:收到后来提交的任务

main:io.run() 返回,执行了 1 个 handler

具體打印先後可能因為線程調度略有變化,但核心流程不變。


3.5.整個執行過程

程序首先創建:

boost::asio::io_context io;

此時:

io_context
↓
没有任何任务

然後:

WorkGuard guard =
    boost::asio::make_work_guard(io);

相當於告訴 io_context:

虽然现在可能没有任务,

但先不要认为整个事件循环已经结束,

以后还可能有人给你提交工作。

接着啓動:

std::thread worker(...);

這個線程會先:

std::this_thread::sleep_for(2s);

所以剛開始的時候,io_context 裏面其實真的沒有任何 handler 可以執行。

此時 main 執行:

io.run();

如果沒有 guard,邏輯會變成:

io.run()
↓
当前没有任何工作
↓
立即返回

但現在有 guard:

io.run()
↓
当前没有 handler
↓
但是 work_guard 仍然有效
↓
说明仍然存在 outstanding work
↓
不能退出
↓
继续阻塞等待

於是 main 就停在:

io.run();

裏面等待。


3.6.2 秒以後發生什麼

worker 睡眠結束:

std::this_thread::sleep_for(2s);

然後執行:

boost::asio::post(
    *io,
    [] {
        std::println(
            "Asio handler:收到后来提交的任务");
    });

post() 相當於:

向 io_context 投遞一個以後要執行的 handler。

於是原本正在:

阻塞等待

的 io.run() 發現:

有新任务来了

然後執行這個 handler。


3.7.guard.reset() 到底做什麼

接下來:

guard->reset();

它並不是:

停止 io_context

也不是:

取消所有任务

更不是:

让 run() 立刻强制返回

它真正表達的是:

我之前登記的那份“佔位工作”現在結束了,你以後不用再為了我強行保持 run() 存活。

可以粗略理解成:

创建 guard

outstanding work:
+1

然後:

guard.reset()

outstanding work:
-1

於是 io_context 又恢復正常退出規則:

如果还有真正的任务
→ 继续处理

如果还有未完成异步操作
→ 继续等待

如果什么都没有了
→ run() 可以返回

所以:

guard.reset();

不是:

命令 run() 退出

而是:

取消“禁止 run() 因没工作而退出”这个限制

這一點非常重要。


3.8.為什麼 post 以後再 reset

本例中:

boost::asio::post(*io, handler);

guard->reset();

順序可以理解成:

先把真正的工作提交进去
↓
然后取消 guard 的占位工作

這時 io_context 中仍然還有:

刚刚 post 的 handler

所以即使 guard 被 reset:

还有真正工作
↓
run() 继续
↓
执行 handler
↓
handler 执行结束
↓
此时已经没有其他工作
↓
run() 返回

完整過程:

创建 io_context
│
├─ 当前没有任务
│
↓
创建 work_guard
│
├─ 人为保持 outstanding work
│
↓
main:io.run()
│
├─ 当前没 handler
│
├─ 但 guard 还在
│
└─ 所以阻塞等待,不退出
│
│
│       worker 睡眠 2 秒
│
│               ↓
│       post(handler)
│               ↓
│       guard.reset()
│
↓
io.run() 收到 handler
│
↓
执行 handler
│
↓
handler 完成
│
├─ guard 已 reset
├─ 没有其他任务
├─ 没有未完成异步操作
│
↓
io.run() 返回

這就是 work_guard 的完整作用。


3.9.如果一直不 reset() 會怎樣

假設:

WorkGuard guard =
    boost::asio::make_work_guard(io);

io.run();

又永遠不:

guard.reset();

那麼即使所有真正任務都已經執行完:

handler 全执行完了

timer 全结束了

socket 也没任务了

io.run() 仍然會認為:

还有一份 outstanding work

所以:

不会正常返回

它會繼續等待以後可能到來的任務。

這正是後台 Asio 線程經常需要的效果。


3.10.reset() 和 io.stop() 不一樣

這兩個很容易混。

guard.reset() 的意思是:

允許 io.run() 在所有真正工作完成以後自然退出。

也就是:

reset()
↓
不再人为保持 io_context 存活
↓
还有任务就继续执行
↓
任务全部完成后 run() 自然返回

而:

io.stop();

更像:

要求這個 io_context 儘快停止運行。

所以簡單理解:

guard.reset()

→ 我不再拦着你退出
→ 剩余工作正常处理
→ 没工作以后自然退出

而:

io.stop()

→ 主动要求事件循环停止

因此如果希望:

尽量优雅地把已经提交的任务处理完再退出

通常更關注:

guard.reset();

如果希望主動讓事件循環停止,則可以使用:

io.stop();

3.11.機械人程序中為什麼經常需要 work_guard

例如你的程序可能是:

main
│
├─ 创建 io_context
│
├─ 创建串口驱动
│
├─ 创建 Asio 通信线程
│
└─ ROS2 开始运行

Asio 通信線程:

io.run();

而真正的異步任務可能是之後才出現:

ROS2 callback
↓
要求发送串口命令
↓
post 到 io_context

或者

设备稍后连接
↓
启动 async_read

或者

以后动态创建 timer

如果 Asio 線程剛啓動時:

暂时没有任何 outstanding work

那麼沒有 guard:

Asio 线程
↓
io.run()
↓
没任务
↓
直接返回
↓
通信线程死掉

使用 guard:

Asio 线程
↓
io.run()
↓
暂时没任务
↓
guard 还在
↓
继续等待
↓
以后任务来了
↓
正常处理

因此可以把 work_guard 理解成:

給長期運行的 Asio 事件循環辦理了一張“先別下班”的憑證。


3.12.make_work_guard() 和 reset() 怎麼記

創建:

auto guard =
    boost::asio::make_work_guard(io);

可以理解成:

告诉 io_context:

“我这里还有未来工作,
即使眼前没任务,你也先别让 run() 下班。”

釋放:

guard.reset();

可以理解成:

告诉 io_context:

“我这边以后不会再依赖你继续保持运行了,
如果其他工作也都结束了,你可以让 run() 返回。”

所以它們不是:

make_work_guard()
→ 创建一个实际任务

reset()
→ 删除一个实际任务

而是:

make_work_guard()
→ 持有一份 outstanding work

reset()
→ 放弃这份 outstanding work

3.13.本示例最重要的結論

沒有 work_guard:

io.run()
↓
暂时没有工作
↓
直接返回

有 work_guard:

io.run()
↓
暂时没有工作
↓
但 guard 仍持有 outstanding work
↓
继续阻塞等待未来任务

釋放 guard:

guard.reset()
↓
不再人为维持 io_context 存活
↓
如果真正工作也全部结束
↓
io.run() 返回

所以一句話記住:

work_guard 不是一個真正要執行的任務,而是一份“還有工作存在”的佔位聲明,用來防止 io.run() 在暫時無任務時提前返回。


4.示例 4:buffer 的基本使用

4.1.程序目標

Boost.Asio 讀寫數據時經常寫:

boost::asio::buffer(data)

本例先看它的基本含義。

4.2.完整代碼

#include <boost/asio.hpp>
#include <array>
#include <print>
#include <string>

int main()
{
    // 程序从 main 函数开始执行,下面的语句会按顺序运行。
    std::string msg = "hello asio";
    auto buf1 = boost::asio::buffer(msg);

    // std::array 是固定长度数组,长度在编译期就确定。
    std::array<char, 128> data;
    auto buf2 = boost::asio::buffer(data);

    std::println("msg.size() = {}", msg.size());
    std::println("buffer(msg).size() = {}", buf1.size());
    std::println("array size = {}", data.size());
    std::println("buffer(data).size() = {}", buf2.size());

    return 0;
}

運行結果:見下方“運行輸出與時間順序”;如果示例涉及定時器、線程、網絡或外部設備,具體時間和順序可能會隨環境略有變化。

4.3.運行輸出

msg.size() = 10
buffer(msg).size() = 10
array size = 128
buffer(data).size() = 128

4.4.本示例需要注意的點

boost::asio::buffer() 通常不會複製數據,它只是生成一個“指向已有內存的 buffer 視圖”。

所以異步讀寫時特別注意:

buffer 指向的原始数据必须活到异步回调完成。

錯誤寫法示意:

void send()
{
    std::string msg = "hello";
    boost::asio::async_write(socket_, boost::asio::buffer(msg), handler);
}

這個 msg 在 send() 返回後就銷燬了,但異步寫可能還沒完成。

正確思路:

把 msg 做成类成员变量
或者用 shared_ptr 管理
或者使用写队列保存待发送数据

4.5.boost::asio::buffer() 可以接收哪些類型

boost::asio::buffer(data) 的作用可以簡單理解為:

把一段連續內存包裝成 Asio 可以使用的 buffer 描述對象。

它本身不會複製數據,主要記錄的是:

内存起始地址
+
内存长度

所以最常見的要求是:

data 最好是一段連續存儲的內存。

常見可直接使用的類型包括:

char data[128];

std::array<char, 128> data;

std::array<std::uint8_t, 128> data;

std::vector<char> data;

std::vector<std::uint8_t> data;

std::string data;

例如:

std::array<std::uint8_t, 256> data;

auto buffer = boost::asio::buffer(data);

或者:

//创建一个名为 data 的 std::vector,里面有 256 个 std::uint8_t 元素。
std::vector<std::uint8_t> data(256);

auto buffer = boost::asio::buffer(data);

字符串也可以:

std::string message = "hello";

auto buffer = boost::asio::buffer(message);

還可以直接指定:

指针 + 长度

例如:

std::uint8_t data[256];

auto buffer =
    boost::asio::buffer(data, 100);

表示只使用:

data 开始位置后的前 100 字节

也可以:

std::uint8_t* data = ...;
std::size_t size = ...;

auto buffer =
    boost::asio::buffer(data, size);

對於串口、網絡等字節流通信,通常優先使用:

std::uint8_t
unsigned char
char
std::byte

例如:

std::array<std::uint8_t, 256> rx_buffer;

這是很常見的接收緩衝區寫法。


像下面這些容器不能簡單看成一整塊連續內存:

std::list
std::deque

因此一般不能像:

std::vector
std::array
std::string

那樣直接作為一整塊連續 buffer 使用。


還要注意 const。

例如:

const std::string message = "hello";

那麼:

boost::asio::buffer(message);

得到的是隻讀 buffer,適合:

发送数据

但不能作為接收緩衝區,因為接收操作需要修改底層內存。

可以簡單理解成:

非 const 数据
→ 可以生成可写 buffer

const 数据
→ 只能生成只读 buffer

最後還有一個非常重要的點:

boost::asio::buffer(data) 不擁有 data 的內存。

例如:

std::string data = "hello";

auto buffer = boost::asio::buffer(data);

這裏 buffer 只是指向 data 的內存。

所以異步操作:

boost::asio::async_write(
    socket,
    boost::asio::buffer(data),
    handler);

完成之前,必須保證:

data 仍然存在

并且底层内存没有失效

因此可以把:

boost::asio::buffer(data)

簡單記成:

把一段連續內存包裝成“地址 + 長度”的 Asio buffer,並不會複製或擁有這段數據。


5.示例 5:類裏封裝 io_context 工作線程

5.1.程序目標

工程裏經常希望 Asio 在單獨線程運行。
本例寫一個最小的 IoThread 類:

  1. 構造時創建 work_guard;
  2. start() 開線程執行 io.run();
  3. stop() 停止線程;
  4. 用 post() 投遞任務驗證效果。

5.2.完整代碼

#include <boost/asio.hpp>
#include <functional>
#include <print>
#include <memory>
#include <string>
#include <thread>

class IoThread
{
public:
    IoThread()
        : guard_(boost::asio::make_work_guard(io_))
    {
    }

    ~IoThread()
    {
        stop();
    }

    void start()
    {
        thread_ = std::thread(std::bind(&IoThread::run, this));
    }

    void stop()
    {
        if (!stopped_)
        {
            stopped_ = true;
            guard_.reset();
            io_.stop();

            if (thread_.joinable())
            {
                thread_.join();
            }
        }
    }

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

private:
    void run()
    {
        std::println("IoThread:io.run() 开始");
        // 启动事件循环,前面注册的异步任务会在这里被调度执行。
        io_.run();
        std::println("IoThread:io.run() 返回");
    }

private:
    // io_context 是 Asio 的事件循环对象,异步任务需要靠它调度。
    boost::asio::io_context io_;
    boost::asio::executor_work_guard<boost::asio::io_context::executor_type> guard_;
    std::thread thread_;
    bool stopped_ = false;
};

void print_task(const std::string& msg)
{
    std::println("执行任务:{}", msg);
}

int main()
{
    // 程序从 main 函数开始执行,下面的语句会按顺序运行。
    IoThread io_thread;
    io_thread.start();

    // post 只是把任务放进队列,真正执行要等 io.run()。
    boost::asio::post(io_thread.io(), std::bind(print_task, std::string("A")));
    boost::asio::post(io_thread.io(), std::bind(print_task, std::string("B")));

    std::this_thread::sleep_for(std::chrono::seconds(1));

    std::println("main:准备停止 IoThread");
    io_thread.stop();

    std::println("main:结束");

    return 0;
}

運行結果:見下方“運行輸出與時間順序”;如果示例涉及定時器、線程、網絡或外部設備,具體時間和順序可能會隨環境略有變化。

5.3.運行輸出與時間順序

程序啓動後,工作線程開始運行:

IoThread:io.run() 开始
执行任务:A
执行任务:B

約 1 秒後主線程停止它:

main:准备停止 IoThread
IoThread:io.run() 返回
main:结束

完整輸出類似:

IoThread:io.run() 开始
执行任务:A
执行任务:B
main:准备停止 IoThread
IoThread:io.run() 返回
main:结束

5.4.本示例需要注意的點

這個類就是以後封裝串口、TCP、UDP 通信線程的雛形。

但注意:

io_.stop();

會讓 io_context 儘快停止,未完成的異步任務可能不會正常完成。

工程裏更細緻的做法是:

先 cancel socket / timer / serial_port
再 reset work_guard
最后等待 run() 返回

6.本節總結

  1. 沒有任務時,io.run() 立刻返回。
  2. post() 可以把普通函數投遞給 io_context 執行。
  3. work_guard 可以防止 io.run() 因為沒任務而退出。
  4. std::bind 可以綁定普通函數,也可以綁定成員函數。
  5. boost::asio::buffer() 通常只是內存視圖,不負責延長數據生命週期。
  6. 異步工程裏最重要的是對象生命週期和 buffer 生命週期。