ASIO：学习一-Aet

Bingliaolong 2026-08-01 16:19 Aet | 隐藏边栏 | 抢沙发 | 2 0

文章评分 1 次，平均分 5.0 ：

同步 `I/O`：服务端

`code`

#include <array>

#include <asio.hpp>

#include <iostream>

using asio::ip::tcp;

// 处理单个客户端连接：同步读、同步写，直到客户端断开

void handle_session(tcp::socket socket) {

try {

for (;;) {

std::array<char, 1024> buf;

asio::error_code ec;

// 同步读取：这里会阻塞当前线程，直到有数据到达或出错

size_t len = socket.read_some(asio::buffer(buf), ec);

if (ec == asio::error::eof) {

std::cout << "客户端断开连接\n";

break; // 对端正常关闭

} else if (ec) {

throw asio::system_error(ec); // 其他错误

}

std::cout << "收到 " << len << " 字节，回显中...\n";

// 同步写回：原样 echo 给客户端

asio::write(socket, asio::buffer(buf, len));

}

} catch (std::exception& e) {

std::cerr << "会话异常: " << e.what() << "\n";

}

int main(int argc, char* argv[]) {

try {

/*if (argc != 2) {

std::cerr << "用法: " << argv[0] << " <port>\n";

return 1;

}*/

// io_context 是所有 asio I/O 对象的核心调度器

// 在纯同步用法里，它主要用来支撑 acceptor / socket 的构造，

// 不需要调用 io_context.run()

asio::io_context io_context;

tcp::acceptor acceptor(io_context, tcp::endpoint(tcp::v4(), 8899));

//std::cout << "服务端启动，监听端口 " << argv[1] << " ...\n";

for (;;) {

tcp::socket socket(io_context);

// 同步 accept：阻塞直到有新连接到来

acceptor.accept(socket);

std::cout << "新客户端已连接: " << socket.remote_endpoint() << "\n";

// 注意：这是单线程同步服务端，同一时刻只能处理一个客户端

// 处理完当前客户端才会回到 accept 等下一个连接

handle_session(std::move(socket));

}

} catch (std::exception& e) {

std::cerr << "异常: " << e.what() << "\n";

}

return 0;

}

同步 `I/O`：客户端

`code`

#include <array>

#include <asio.hpp>

#include <iostream>

#include <string>

using asio::ip::tcp;

int main(int argc, char* argv[]) {

try {

/*if (argc != 3) {

std::cerr << "用法: " << argv[0] << " <host> <port>\n";

return 1;

}*/

asio::io_context io_context;

// resolver 负责把 host/port 解析成一个或多个 endpoint

// 这一步在同步模式下同样是阻塞调用

tcp::resolver resolver(io_context);

tcp::resolver::results_type endpoints = resolver.resolve("127.0.0.1", "8899");

tcp::socket socket(io_context);

// asio::connect 会依次尝试 endpoints 里的每一个地址

// 直到连接成功或全部失败

asio::connect(socket, endpoints);

std::cout << "已连接到服务端 " << socket.remote_endpoint() << "\n";

std::cout << "输入内容后回车发送，输入 quit 退出\n";

std::string line;

while (std::getline(std::cin, line)) {

if (line == "quit")

break;

// 同步写：阻塞直到数据全部写入 socket 缓冲区

asio::write(socket, asio::buffer(line + "\n"));

std::array<char, 1024> buf;

asio::error_code ec;

// 同步读：阻塞直到服务端有数据回来

size_t len = socket.read_some(asio::buffer(buf), ec);

if (ec == asio::error::eof) {

std::cout << "服务端已关闭连接\n";

break;

} else if (ec) {

throw asio::system_error(ec);

}

std::cout << "服务端回显: " << std::string(buf.data(), len);

}

} catch (std::exception& e) {

std::cerr << "异常: " << e.what() << "\n";

}

return 0;

}

异步`I/O`：服务端

`code`

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

#include <array>

#include <asio.hpp>

#include <iostream>

#include <memory>

using asio::ip::tcp;

// -----------------------------------------------------------------------

// session：代表一个客户端连接的生命周期

// 必须继承 enable_shared_from_this，因为异步操作的回调是"未来某个时间点"

// 才会被调用的，这段时间内 session 本身绝不能被销毁——

// 靠 shared_ptr 在每个异步操作的回调里"续命"，直到没有回调再持有它为止

// -----------------------------------------------------------------------

class session : public std::enable_shared_from_this<session> {

public:

explicit session(tcp::socket socket) : socket_(std::move(socket)) {}

void start() { do_read(); }

private:

void do_read() {

auto self(shared_from_this()); // 延长生命周期，供 lambda 捕获

// 发起一次异步读操作：

// - 立即返回（不阻塞），真正的数据到达后由 io_context::run() 所在的线程调用回调

// - async_read_some 只保证"读到至少一个字节"，不保证读满 buffer

socket_.async_read_some(asio::buffer(data_),

[this, self](std::error_code ec, std::size_t length) {

if (!ec) {

std::cout << "收到 " << length << " 字节，准备回显\n";

do_write(length);

} else if (ec == asio::error::eof) {

std::cout << "客户端断开连接\n";

} else {

std::cerr << "读错误: " << ec.message() << "\n";

}

// 什么都不做时，self 这个 shared_ptr 在 lambda 结束后释放，

// 如果这是最后一个引用，session 就会被销毁

});

}

void do_write(std::size_t length) {

auto self(shared_from_this());

// asio::async_write 保证把 length 个字节全部写完才回调

// （区别于 async_write_some，只保证写入"至少一部分"）

asio::async_write(socket_, asio::buffer(data_, length),

[this, self](std::error_code ec, std::size_t /*bytes_transferred*/) {

if (!ec) {

// 写完后，继续发起下一次读，形成"读-写-读-写"的异步循环

do_read();

} else {

std::cerr << "写错误: " << ec.message() << "\n";

}

});

}

tcp::socket socket_;

std::array<char, 1024> data_;

};

// -----------------------------------------------------------------------

// server：负责持续 async_accept 新连接

// -----------------------------------------------------------------------

class server {

public:

server(asio::io_context& io_context, unsigned short port) :

acceptor_(io_context, tcp::endpoint(tcp::v4(), port)) {

do_accept();

}

private:

void do_accept() {

// 异步接受连接：立即返回，连接到来时回调被触发

acceptor_.async_accept([this](std::error_code ec, tcp::socket socket) {

if (!ec) {

std::cout << "新客户端已连接: " << socket.remote_endpoint() << "\n";

// 为这个连接创建一个 session 并启动它的读写循环

std::make_shared<session>(std::move(socket))->start();

} else {

std::cerr << "accept 错误: " << ec.message() << "\n";

}

// 无论这次 accept 成功与否，都要继续挂起下一次 accept，

// 否则服务端只能接受一个连接就"停摆"了

do_accept();

});

}

tcp::acceptor acceptor_;

};

int main(int argc, char* argv[]) {

try {

/*if (argc != 2) {

std::cerr << "用法: " << argv[0] << " <port>\n";

return 1;

}*/

asio::io_context io_context;

server s(io_context, static_cast<unsigned short>(8899));

//server s(io_context, static_cast<unsigned short>(std::atoi(argv[1])));

//std::cout << "异步服务端启动，监听端口 " << argv[1] << " ...\n";

// 关键：run() 会阻塞在这里，不断从内部事件队列取出"就绪的回调"并执行

// 前面所有 async_xxx 发起的操作，都是在这里被真正触发和调度的

io_context.run();

} catch (std::exception& e) {

std::cerr << "异常: " << e.what() << "\n";

}

return 0;

}

异步`I/O`：客户端

`code`

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

#include <array>

#include <asio.hpp>

#include <deque>

#include <iostream>

#include <memory>

#include <string>

#include <thread>

using asio::ip::tcp;

class client {

public:

client(asio::io_context& io_context, const std::string& host, const std::string& port) :

io_context_(io_context), socket_(io_context), resolver_(io_context) {

do_resolve(host, port);

}

// 供外部线程（比如从标准输入读取用户输入的线程）投递要发送的消息

void write(const std::string& msg) {

// post 把"往 write_queue_ 里塞数据 + 发起 async_write"这件事

// 转交给 io_context 所在的线程去做，避免多线程直接操作 socket_

asio::post(io_context_, [this, msg]() {

bool writing_in_progress = !write_queue_.empty();

write_queue_.push_back(msg);

if (!writing_in_progress) {

do_write();

}

});

}

void close() {

asio::post(io_context_, [this]() { socket_.close(); });

}

private:

void do_resolve(const std::string& host, const std::string& port) {

// 异步域名解析：立即返回，解析完成后回调被触发

resolver_.async_resolve(host, port,

[this](std::error_code ec, tcp::resolver::results_type endpoints) {

if (!ec) {

do_connect(endpoints);

} else {

std::cerr << "解析失败: " << ec.message() << "\n";

}

});

}

void do_connect(const tcp::resolver::results_type& endpoints) {

// asio::async_connect 会依次尝试 endpoints，

// 直到某一个连接成功，或全部失败才回调

asio::async_connect(socket_, endpoints,

[this](std::error_code ec, const tcp::endpoint& /*ep*/) {

if (!ec) {

std::cout << "已连接到服务端 " << socket_.remote_endpoint() << "\n";

do_read();

} else {

std::cerr << "连接失败: " << ec.message() << "\n";

}

});

}

void do_read() {

socket_.async_read_some(

asio::buffer(read_buf_), [this](std::error_code ec, std::size_t length) {

if (!ec) {

std::cout << "服务端回显: " << std::string(read_buf_.data(), length);

do_read(); // 继续等待下一次数据

} else if (ec == asio::error::eof) {

std::cout << "服务端已关闭连接\n";

} else {

std::cerr << "读错误: " << ec.message() << "\n";

}

});

}

void do_write() {

asio::async_write(socket_, asio::buffer(write_queue_.front()),

// ---------------------------------------------------------

// 下面这行是 lambda 写法；如果要改成 boost::bind 风格

// （老代码库里很常见，尤其是 C++03/C++11 早期项目），等价写法是：

// asio::async_write(socket_, asio::buffer(write_queue_.front()),

// boost::bind(&client::handle_write, this,

// asio::placeholders::error));

// 其中 handle_write 需要单独定义成一个成员函数：

// void handle_write(const std::error_code& ec) { ... }

// 两者本质等价：boost::bind 生成的是一个函数对象，

// lambda 是编译器帮你生成同样效果的匿名函数对象，

// 现代 C++ 项目里基本都用 lambda 替代 boost::bind 了。

// ---------------------------------------------------------

[this](std::error_code ec, std::size_t /*bytes_transferred*/) {

if (!ec) {

write_queue_.pop_front();

if (!write_queue_.empty()) {

do_write(); // 队列里还有待发送的消息，继续发

}

} else {

std::cerr << "写错误: " << ec.message() << "\n";

socket_.close();

}

});

}

asio::io_context& io_context_;

tcp::socket socket_;

tcp::resolver resolver_;

std::array<char, 1024> read_buf_;

std::deque<std::string> write_queue_; // 发送队列，避免并发写冲突

};

int main(int argc, char* argv[]) {

try {

/*if (argc != 3) {

std::cerr << "用法: " << argv[0] << " <host> <port>\n";

return 1;

}*/

asio::io_context io_context;

//client c(io_context, argv[1], argv[2]);

client c(io_context, "127.0.0.1", "8899");

// 用另一个线程跑 io_context::run()，主线程专心读用户输入

// 这是异步模型的典型用法：I/O 线程和"业务/UI"线程分离

std::thread io_thread([&io_context]() { io_context.run(); });

std::string line;

std::cout << "输入内容后回车发送，输入 quit 退出\n";

while (std::getline(std::cin, line)) {

if (line == "quit")

break;

c.write(line + "\n");

}

c.close();

io_thread.join();

} catch (std::exception& e) {

std::cerr << "异常: " << e.what() << "\n";

}

return 0;

}

定时器

`code`

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

#include <asio.hpp>

#include <chrono>

#include <iostream>

#include <memory>

using asio::steady_timer;

// -----------------------------------------------------------------------

// 示例 1：同步等待（对照你之前学的同步 I/O）

// -----------------------------------------------------------------------

void demo_sync_wait(asio::io_context& io_context) {

std::cout << "[示例1] 同步等待 2 秒...\n";

steady_timer timer(io_context);

// expires_after 设置"从现在开始，多久之后到期"

// 这是相对时间；如果要指定绝对时间点，用 expires_at(time_point)

timer.expires_after(std::chrono::seconds(2));

// 同步 wait：阻塞当前线程，直到定时器到期

timer.wait();

std::cout << "[示例1] 2 秒到，继续执行\n\n";

}

// -----------------------------------------------------------------------

// 示例 2：异步一次性等待 + lambda 回调

// -----------------------------------------------------------------------

void demo_async_wait_once(asio::io_context& io_context) {

std::cout << "[示例2] 发起异步等待 1 秒（不阻塞，立即返回）...\n";

// 注意：timer 必须活到回调触发的那一刻，这里用 shared_ptr 延长生命周期

// （如果 timer 是局部变量，函数返回后就析构，回调触发时会访问已销毁对象）

auto timer = std::make_shared<steady_timer>(io_context);

timer->expires_after(std::chrono::seconds(1));

timer->async_wait([timer](std::error_code ec) // 捕获 timer 自身，保证回调触发前它不被销毁

{

if (!ec) {

std::cout << "[示例2] 1 秒到，回调被触发\n\n";

} else {

// 定时器被 cancel() 时，ec 会是 asio::error::operation_aborted

std::cout << "[示例2] 定时器被取消: " << ec.message() << "\n\n";

}

});

std::cout << "[示例2] 主流程继续往下走，不会被定时器阻塞\n";

}

// -----------------------------------------------------------------------

// 示例 3：周期性定时器

// asio 没有"重复定时器"这个 API，标准做法是：

// 在回调里重新设置 expires_after 并再次 async_wait，形成"自我重新武装"的循环

// -----------------------------------------------------------------------

class periodic_printer : public std::enable_shared_from_this<periodic_printer> {

public:

periodic_printer(asio::io_context& io_context, std::chrono::milliseconds interval) :

timer_(io_context), interval_(interval), count_(0) {}

void start() { do_wait(); }

void stop() {

// cancel() 会让所有挂起的 async_wait 立刻以 operation_aborted 触发回调

timer_.cancel();

}

private:

void do_wait() {

auto self(shared_from_this());

timer_.expires_after(interval_);

timer_.async_wait([this, self](std::error_code ec) {

if (ec) {

// 被取消或出错，不再继续重新武装定时器，循环自然终止

std::cout << "[示例3] 周期定时器停止: " << ec.message() << "\n";

return;

}

++count_;

std::cout << "[示例3] 第 " << count_ << " 次心跳\n";

if (count_ < 5) {

do_wait(); // 关键：在回调里重新发起下一次等待

} else {

std::cout << "[示例3] 已完成 5 次心跳，主动停止\n\n";

}

});

}

steady_timer timer_;

std::chrono::milliseconds interval_;

int count_;

};

// -----------------------------------------------------------------------

// 示例 4：手动 cancel() 一个正在等待的定时器

// -----------------------------------------------------------------------

void demo_cancel(asio::io_context& io_context) {

std::cout << "[示例4] 演示手动取消定时器\n";

auto timer = std::make_shared<steady_timer>(io_context);

timer->expires_after(std::chrono::seconds(10)); // 故意设置一个很长的等待

timer->async_wait([](std::error_code ec) {

if (ec == asio::error::operation_aborted) {

std::cout << "[示例4] 定时器在到期前被取消了（符合预期）\n\n";

} else if (!ec) {

std::cout << "[示例4] 定时器正常到期（不符合预期，说明取消失败）\n\n";

}

});

// 100 毫秒后，用另一个定时器触发 cancel() ——

// 模拟"业务逻辑决定提前中断等待"的场景

auto canceller = std::make_shared<steady_timer>(io_context);

canceller->expires_after(std::chrono::milliseconds(100));

canceller->async_wait([timer, canceller](std::error_code) { timer->cancel(); });

}

int main() {

try {

asio::io_context io_context;

// 示例1是同步的，直接跑，不需要 io_context.run()

demo_sync_wait(io_context);

// 示例2、3、4 都是异步操作，需要 io_context.run() 才能真正被调度执行

demo_async_wait_once(io_context);

auto printer = std::make_shared<periodic_printer>(io_context, std::chrono::milliseconds(500));

printer->start();

demo_cancel(io_context);

std::cout << "[main] 调用 io_context.run()，开始处理所有挂起的异步操作...\n\n";

// run() 会一直阻塞，直到所有异步操作都完成（没有更多待处理的 handler）

io_context.run();

std::cout << "[main] 所有异步操作已完成，io_context.run() 返回\n";

} catch (std::exception& e) {

std::cerr << "异常: " << e.what() << "\n";

}

return 0;

}

线程池

`code`

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

// 多线程运行 io_context::run()：体会"线程池驱动事件循环"

// 涵盖：

// 1. 用多个线程一起调用 io_context.run()，观察 handler 被不同线程执行

// 2. 用 work_guard 防止 run() 提前返回

// 3. 故意制造一个"无保护共享变量"的数据竞争，验证多线程下真的存在竞态

// 4. 用简单的互斥锁修复，为下一步学 strand 做铺垫

// 编译: g++ -std=c++17 -O2 thread_pool_io_context.cpp -o thread_pool_io_context

// -I<asio_include_path> -lpthread 运行: ./thread_pool_io_context

#include <asio.hpp>

#include <iostream>

#include <mutex>

#include <sstream>

#include <thread>

#include <vector>

// -----------------------------------------------------------------------

// 工具函数：把线程 id 转成字符串，方便打印观察"是哪个线程执行了这个 handler"

// -----------------------------------------------------------------------

std::string tid() {

std::ostringstream oss;

oss << std::this_thread::get_id();

return oss.str();

}

// 用来串行化 std::cout 输出的锁（仅用于打印整洁，不代表业务逻辑正确性）

std::mutex cout_mutex;

void safe_print(const std::string& msg) {

std::lock_guard<std::mutex> lock(cout_mutex);

std::cout << msg << "\n";

}

// -----------------------------------------------------------------------

// 示例 1：多个 post() 任务被"线程池"并发执行

// -----------------------------------------------------------------------

void demo_concurrent_handlers(asio::io_context& io_context) {

safe_print("\n===== 示例1：多个任务被线程池并发执行 =====");

for (int i = 0; i < 8; ++i) {

// post() 把一个"任务"塞进 io_context 的任务队列

// 具体由哪个线程执行、什么时候执行，取决于哪个线程的 run() 先抢到它

asio::post(io_context, [i]() {

safe_print("任务 " + std::to_string(i) + " 正在线程 " + tid() + " 中执行");

// 模拟一点耗时工作，让任务有机会被不同线程"抢"到

std::this_thread::sleep_for(std::chrono::milliseconds(50));

});

}

// -----------------------------------------------------------------------

// 示例 2：没有保护的共享计数器 —— 制造真实的数据竞争

// -----------------------------------------------------------------------

long unsafe_counter = 0;

void demo_race_condition(asio::io_context& io_context) {

safe_print("\n===== 示例2：无保护共享变量的数据竞争 =====");

const int N = 100000;

for (int i = 0; i < N; ++i) {

asio::post(io_context, []() {

// 危险：多个线程同时执行到这里，++ 不是原子操作

// (读取 -> +1 -> 写回，三步之间可能被其他线程打断)

++unsafe_counter;

});

}

safe_print("已投递 " + std::to_string(N) + " 个自增任务（预期最终结果应为 " + std::to_string(N) +

"，多线程下大概率会更小）");

}

// -----------------------------------------------------------------------

// 示例 3：用 mutex 保护共享变量（正确，但要付出锁开销；下次学 strand 会有更优雅的方式）

// -----------------------------------------------------------------------

long safe_counter = 0;

std::mutex counter_mutex;

void demo_mutex_fix(asio::io_context& io_context) {

safe_print("\n===== 示例3：用 mutex 保护后的正确自增 =====");

const int N = 100000;

for (int i = 0; i < N; ++i) {

asio::post(io_context, []() {

std::lock_guard<std::mutex> lock(counter_mutex);

++safe_counter;

});

}

int main() {

try {

asio::io_context io_context;

// -------------------------------------------------------------

// work_guard 的作用：

// 如果队列一度空了（比如所有 demo 函数还没来得及投递任务），

// run() 会立刻返回并让线程退出。work_guard 存在期间，

// 即使队列暂时为空，run() 也不会返回，会继续等待新任务。

// 相当于告诉 io_context："先别退出，后面还有活要干"。

// -------------------------------------------------------------

auto work_guard = asio::make_work_guard(io_context);

// 投递本示例要跑的所有任务

demo_concurrent_handlers(io_context);

demo_race_condition(io_context);

demo_mutex_fix(io_context);

// -------------------------------------------------------------

// 核心：创建一个线程池，每个线程都调用 io_context.run()

// 这些线程会"抢"任务队列里的 handler 来执行——

// 这就是"线程池驱动事件循环"的本质：

// io_context 内部的任务队列是线程安全的，

// 谁先调用 run() 里的取任务逻辑，谁就执行下一个 handler

// -------------------------------------------------------------

unsigned int thread_count = std::thread::hardware_concurrency();

if (thread_count == 0)

thread_count = 4;

safe_print("启动 " + std::to_string(thread_count) + " 个线程运行 io_context.run()");

std::vector<std::thread> threads;

for (unsigned int i = 0; i < thread_count; ++i) {

threads.emplace_back([&io_context]() { io_context.run(); });

}

// 主线程稍等一下，让上面投递的任务基本执行完

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

// 释放 work_guard：告诉 io_context "没有更多长期挂起的工作了"

// 此时一旦任务队列被消费完，所有线程的 run() 就会自然返回

work_guard.reset();

for (auto& t : threads) {

t.join();

}

safe_print("\n===== 结果对比 =====");

safe_print("unsafe_counter 最终值 = " + std::to_string(unsafe_counter) +

"（理论期望 100000，若数值更小，说明确实发生了数据竞争丢失自增）");

safe_print("safe_counter 最终值 = " + std::to_string(safe_counter) +

"（用 mutex 保护后应精确等于 100000）");

safe_print("\n所有线程已退出，io_context 运行结束");

} catch (std::exception& e) {

std::cerr << "异常: " << e.what() << "\n";

}

return 0;

}

`strand`

概述

io_context::run() 可以被多个线程同时调用（线程池模式）。这时候多个 handler（回调）可能在不同线程上并发执行
如果这些 handler 访问了共享数据（比如一个 std::map、一个连接列表、一个计数器），就会有数据竞争

最直觉的解法

"每个共享数据配一把 mutex"
但这样做有几个隐藏代价：
1. 锁的粒度难控制：细粒度锁容易死锁（尤其是异步回调里又发起新的异步操作，持锁跨越异步边界几乎必然出问题）；粗粒度锁又退化成单线程性能
2. 异步回调里加锁本身很危险：如果在持锁期间调用了另一个可能同步触发同一把锁的操作，就会死锁。而 ASIO 的 handler 经常是链式调用的
3. 锁不解决"逻辑顺序"问题：即使数据不竞争了，你可能仍然需要"这几个操作必须按顺序发生"，普通 mutex 只保证互斥，不保证顺序性

`strand`的思路

现在叫 asio::strand<Executor>
1. 它不是"保护数据"，而是保证所有绑定到同一个 strand 的 handler 永远不会并发执行、且按 post 的顺序执行（不是抢锁，是排队执行）
2. 这样你回到了单线程编程模型，不需要锁，也不会死锁

`code`：不用 `strand`

有数据竞争

#include <asio.hpp>

#include <iostream>

#include <thread>

#include <vector>

asio::io_context io;

int shared_counter = 0; // 共享数据，没有保护

void unsafe_increment(int id) {

for (int i = 0; i < 100000; ++i) {

++shared_counter; // 数据竞争！多线程下结果不可预测

}

int main() {

// 用 io_context::post 模拟多个"异步任务"

for (int i = 0; i < 4; ++i) {

asio::post(io, [i]() { unsafe_increment(i); });

}

// 4 个线程一起 run，handler 可能被不同线程执行

std::vector<std::thread> pool;

for (int i = 0; i < 4; ++i)

pool.emplace_back([]() { io.run(); });

for (auto& t : pool) t.join();

std::cout << "counter = " << shared_counter << std::endl; // 大概率 < 400000

}

`code`：无脑加锁

能用，但有代价

#include <mutex>

std::mutex mtx;

int shared_counter = 0;

void locked_increment() {

for (int i = 0; i < 100000; ++i) {

std::lock_guard<std::mutex> lock(mtx);

++shared_counter;

}

这个能解决问题，但想象一下：
1. 如果 shared_counter 的更新逻辑里，还要根据结果发起另一个异步操作（比如往 socket 写数据），你就得在持锁期间调用 async_write，或者解锁后再调用
2. 这时候"操作的顺序性"谁来保证？
3. 两个线程可能都读到"该发送"的状态，各自解锁后都发起了写操作，顺序全乱了
4. mutex 只管"同一时刻只有一个人在改数据"，不管"事情发生的先后逻辑"

`code：strand`

ASIO 方式
关键点：
1. 所有对 shared_counter 的修改都被 post 到同一个 strand_ 上

#include <asio.hpp>

#include <iostream>

#include <thread>

#include <vector>

asio::io_context io;

asio::strand<asio::io_context::executor_type> strand_ = asio::make_strand(io);

int shared_counter = 0; // 不需要 mutex！

void strand_increment(int id) {

for (int i = 0; i < 100000; ++i) {

// 把每次自增包装成一个 handler，post 到 strand 上

asio::post(strand_, []() {

++shared_counter; // 保证不会和其他 strand 上的 handler 并发执行

});

}

int main() {

for (int i = 0; i < 4; ++i) {

asio::post(io, [i]() { strand_increment(i); });

}

std::vector<std::thread> pool;

for (int i = 0; i < 4; ++i)

pool.emplace_back([]() { io.run(); });

for (auto& t : pool) t.join();

std::cout << "counter = " << shared_counter << std::endl; // 稳定等于 400000

}

ASIO 保证：
1. 绑定同一个 strand 的多个 handler，不会被两个线程同时执行（哪怕你有 8 个 worker 线程在 io.run()
2. 它们会严格按照 post 的顺序依次执行（不是加锁抢占，是排队）

`strand<Executor>`

`strand`的原理

strand 内部确实用了一把 mutex，但这把锁只保护"一个链表"的几次指针操作，从不跨越 handler 的执行过程

1 2	Executor executor_; // 底层真正执行任务的 executor（比如 io_context::executor_type） implementation_type impl_; // 指向共享状态的句柄

typedef detail::strand_executor_service::implementation_type

implementation_type;

typedef shared_ptr<strand_impl> implementation_type;

class strand_impl

{

public:

ASIO_DECL ~strand_impl();

private:

friend class strand_executor_service;

// Mutex to protect access to internal data.

#if defined(ASIO_HAS_STD_ATOMIC_WAIT) \

|| defined(ASIO_HAS_FUTEX)

slim_mutex mutex_;

void lock_mutex()

{

mutex_.lock();

}

void unlock_mutex()

{

mutex_.unlock();

}

#else // defined(ASIO_HAS_STD_ATOMIC_WAIT)

// || defined(ASIO_HAS_FUTEX)

mutex* mutex_;

void lock_mutex()

{

mutex_->lock();

}

void unlock_mutex()

{

mutex_->unlock();

}

#endif // defined(ASIO_HAS_STD_ATOMIC_WAIT)

// || defined(ASIO_HAS_FUTEX)

// Indicates whether the strand is currently "locked" by a handler. This

// means that there is a handler upcall in progress, or that the strand

// itself has been scheduled in order to invoke some pending handlers.

bool locked_;

// Indicates that the strand has been shut down and will accept no further

// handlers.

bool shutdown_;

// The handlers that are waiting on the strand but should not be run until

// after the next time the strand is scheduled. This queue must only be

// modified while the mutex is locked.

op_queue<scheduler_operation> waiting_queue_;

// The handlers that are ready to be run. Logically speaking, these are the

// handlers that hold the strand's lock. The ready queue is only modified

// from within the strand and so may be accessed without locking the mutex.

op_queue<scheduler_operation> ready_queue_;

// Pointers to adjacent handle implementations in linked list.

strand_impl* next_;

strand_impl* prev_;

// The strand service in where the implementation is held.

strand_executor_service* service_;

};

impl 是从哪来的
1. use_service 是 ASIO 的 service 机制
2. 每个 execution_context（比如 io_context）内部维护一个全局唯一的 strand_executor_service 实例（单例，按 context 生命周期管理）

template <typename InnerExecutor>

static implementation_type create_implementation(const InnerExecutor& ex,

constraint_t<

can_query<InnerExecutor, execution::context_t>::value

> = 0)

{

return use_service<detail::strand_executor_service>(

asio::query(ex, execution::context)).create_implementation();

}

示例：`TCP` 连接管理器

这是 strand 真正大放异彩的场景——异步操作链中间夹杂着共享状态：

class ConnectionManager {

public:

ConnectionManager(asio::io_context& io)

: strand_(asio::make_strand(io)) {}

// 从任意线程调用都安全

void add_connection(std::shared_ptr<Session> session) {

asio::post(strand_, [this, session]() {

connections_.push_back(session); // 不需要锁

});

}

void broadcast(const std::string& msg) {

asio::post(strand_, [this, msg]() {

for (auto& conn : connections_) {

// 即使这里触发新的异步操作，也不会和

// add_connection 的 handler 产生竞争

conn->async_send(msg);

}

});

}

private:

asio::strand<asio::io_context::executor_type> strand_;

std::vector<std::shared_ptr<Session>> connections_; // 只被 strand 上的 handler 访问

};

协程封装

`C++20 awaitable` + `co_spawn`

awaitable<void> 是一个返回类型
1. 本质是编译器根据 co_await/co_return 关键字生成的一个状态机对象
use_awaitable 是一个 completion token
1. 它告诉 async_read_some 这类函数："不要走回调风格，返回一个 awaitable 给我"
co_spawn(io, coroutine, detached) 才是真正"点火"的地方
1. 它把协程包装成一个 awaitable，然后post 到 io_context 上执行，第一次 resume 由 io_context 的某个线程发起

#include <asio.hpp>

#include <asio/awaitable.hpp>

#include <asio/co_spawn.hpp>

#include <asio/detached.hpp>

#include <asio/use_awaitable.hpp>

#include <iostream>

using asio::ip::tcp;

using asio::awaitable;

using asio::co_spawn;

using asio::detached;

using asio::use_awaitable;

// 处理单个连接的协程

awaitable<void> echo(tcp::socket socket)

{

try {

char data[1024];

for (;;) {

// co_await 会在这里"挂起"当前协程，

// 直到数据读到为止，期间不占用线程

std::size_t n = co_await socket.async_read_some(

asio::buffer(data), use_awaitable);

co_await asio::async_write(

socket, asio::buffer(data, n), use_awaitable);

}

} catch (std::exception& e) {

std::cerr << "echo exception: " << e.what() << "\n";

}

// 监听协程

awaitable<void> listener(asio::io_context& io, unsigned short port)

{

tcp::acceptor acceptor(io, {tcp::v4(), port});

for (;;) {

tcp::socket socket = co_await acceptor.async_accept(use_awaitable);

// 把新连接的协程"丢出去"独立运行，不等它完成

// detached 意味着不关心它的结果/异常（生产环境通常要包一层异常处理）

co_spawn(io, echo(std::move(socket)), detached);

}

int main()

{

asio::io_context io(1); // 单线程也能跑协程，因为协程本身不依赖多线程

co_spawn(io, listener(io, 55555), detached);

io.run();

}

绑定 `strand` 的协程

如果多个协程之间要共享状态，直接把协程绑定到同一个 strand 上，比手写锁简单得多
co_await asio::this_coro::executor
1. 拿到的是当前协程所绑定的 executor（这里是 strand_ex），所以协程内部发起的每一个异步操作、每一次 resume，都会被约束在同一个 strand 上，逻辑上等价于我们之前讲的"接力棒"机制
2. 只是这次接力棒传递的是协程的挂起点，而不是普通 handler

#include <asio/strand.hpp>

using strand_t = asio::strand<asio::io_context::executor_type>;

int shared_counter = 0;

awaitable<void> worker(int id)

{

for (int i = 0; i < 100000; ++i) {

++shared_counter; // 因为整个协程都跑在同一个 strand 上，天然安全

co_await asio::post(

co_await asio::this_coro::executor, use_awaitable); // 让出一下，模拟异步点

}

std::cout << "worker " << id << " done\n";

}

int main()

{

asio::io_context io;

auto strand_ex = asio::make_strand(io);

for (int i = 0; i < 4; ++i) {

// 注意这里传的 executor 是 strand_ex，不是 io.get_executor()

co_spawn(strand_ex, worker(i), detached);

}

std::vector<std::thread> pool;

for (int i = 0; i < 4; ++i)

pool.emplace_back([&io]() { io.run(); });

for (auto& t : pool) t.join();

std::cout << "counter = " << shared_counter << "\n"; // 稳定等于 400000

}

`Stackful` 协程：`asio::spawn`

这是 C++20 之前（甚至现在很多老代码库仍在用）的写法，基于 Boost.Context 实现真正的"可切换栈"，代码看起来完全同步，没有 co_await：
（旧式写法，yield_context）

#include <asio.hpp>

#include <asio/spawn.hpp>

#include <iostream>

using asio::ip::tcp;

void echo(asio::yield_context yield, tcp::socket socket)

{

try {

char data[1024];

for (;;) {

// 没有 co_await，看起来完全是同步代码

// 但底层会切换到调度器的栈，让出执行权，等 IO 完成再切回来

std::size_t n = socket.async_read_some(

asio::buffer(data), yield);

asio::async_write(

socket, asio::buffer(data, n), yield);

}

} catch (std::exception& e) {

std::cerr << "echo exception: " << e.what() << "\n";

}

void listener(asio::yield_context yield, asio::io_context& io, unsigned short port)

{

tcp::acceptor acceptor(io, {tcp::v4(), port});

for (;;) {

tcp::socket socket = acceptor.async_accept(yield);

// spawn 新的 stackful 协程去处理这个连接

asio::spawn(io,

[&io, socket = std::move(socket)](asio::yield_context yield) mutable {

echo(yield, std::move(socket));

[](std::exception_ptr e) { if (e) std::rethrow_exception(e); });

}

int main()

{

asio::io_context io(1);

asio::spawn(io,

[&io](asio::yield_context yield) { listener(yield, io, 55555); },

[](std::exception_ptr e) { if (e) std::rethrow_exception(e); });

io.run();

}

`awaitable`和`spawn`


	`awaitable (C++20)`	`stackful` (`spawn`)
挂起机制	编译器生成状态机，函数栈帧被"拆解"存到堆上	真实的栈切换（类似线程切换，但是用户态），整个调用栈被完整保留
能否在深层嵌套函数里挂起	只有标记为 `awaitable<T>` 的函数能 `co_await`，不能在普通函数里挂起	可以在任意深度的普通函数调用链里挂起（只要 `yield_context` 一路传下去或者存在协程里）
开销	更轻量（无独立栈，编译期优化空间大）	每个协程要分配一个栈（默认几十 `KB` 到 `1MB` 级别），量大时内存开销明显
依赖	纯语言特性（`C++20` 编译器支持即可）	依赖 `Boost.Context`（汇编级别的上下文切换），或者 `ucontext`
现状	ASIO 现在主推的方式	较老，仍在维护，但新项目一般不再首选

`echo server`

单线程

特点
1. 只有一个线程调用 io_context.run()，事件循环全在这个线程里跑
2. 所有并发连接靠“回调套回调”实现，没有数据竞争问题（因为根本没有并发执行）
3. 每个 Session 用 shared_from_this 延长生命周期，防止对象在异步操作完成前被销毁

#include <asio.hpp>

#include <iostream>

#include <memory>

using asio::ip::tcp;

class Session : public std::enable_shared_from_this<Session> {

public:

explicit Session(tcp::socket socket) : socket_(std::move(socket)) {}

void start() { do_read(); }

private:

void do_read() {

auto self(shared_from_this());

socket_.async_read_some(asio::buffer(data_, max_length),

[this, self](asio::error_code ec, std::size_t length) {

if (!ec) {

do_write(length);

} else if (ec != asio::error::eof) {

std::cerr << "read error: " << ec.message() << "\n";

}

});

}

void do_write(std::size_t length) {

auto self(shared_from_this());

asio::async_write(socket_, asio::buffer(data_, length),

[this, self](asio::error_code ec, std::size_t /*length*/) {

if (!ec) {

do_read(); // 继续读下一轮，形成读->写->读->写的回调链

} else {

std::cerr << "write error: " << ec.message() << "\n";

}

});

}

tcp::socket socket_;

enum { max_length = 1024 };

char data_[max_length];

};

class Server {

public:

Server(asio::io_context& io_context, unsigned short port)

: acceptor_(io_context, tcp::endpoint(tcp::v4(), port)) {

do_accept();

}

private:

void do_accept() {

acceptor_.async_accept(

[this](asio::error_code ec, tcp::socket socket) {

if (!ec) {

std::make_shared<Session>(std::move(socket))->start();

}

do_accept(); // 继续接受下一个连接

});

}

tcp::acceptor acceptor_;

};

int main() {

try {

unsigned short port = 8888;

asio::io_context io_context;

Server server(io_context, port);

std::cout << "[single-thread] echo server listening on port " << port << "\n";

io_context.run(); // 单线程独自跑事件循环，直到没有未完成的异步任务

} catch (std::exception& e) {

std::cerr << "Exception: " << e.what() << "\n";

}

return 0;

}

多线程

和单线程版相比的关键变化
1. main() 里起了多个线程共同调用 io_context.run() —— 真正的并发执行
2. 一旦多线程共享 io_context，同一个 socket 上的读/写回调可能被不同线程调度，必须用 strand 把同一个 Session 的操作串行化，否则会有数据竞争
3. 回调风格本身没变，只是多包了一层 bind_executor(strand_, ...)

#include <asio.hpp>

#include <iostream>

#include <memory>

#include <thread>

#include <vector>

using asio::ip::tcp;

class Session : public std::enable_shared_from_this<Session> {

public:

explicit Session(tcp::socket socket)

: socket_(std::move(socket)),

strand_(asio::make_strand(socket_.get_executor())) {}

void start() { do_read(); }

private:

void do_read() {

auto self(shared_from_this());

socket_.async_read_some(asio::buffer(data_, max_length),

asio::bind_executor(strand_,

[this, self](asio::error_code ec, std::size_t length) {

if (!ec) {

do_write(length);

} else if (ec != asio::error::eof) {

std::cerr << "read error: " << ec.message() << "\n";

}

}));

}

void do_write(std::size_t length) {

auto self(shared_from_this());

asio::async_write(socket_, asio::buffer(data_, length),

asio::bind_executor(strand_,

[this, self](asio::error_code ec, std::size_t /*length*/) {

if (!ec) {

do_read();

} else {

std::cerr << "write error: " << ec.message() << "\n";

}

}));

}

tcp::socket socket_;

asio::strand<tcp::socket::executor_type> strand_;

enum { max_length = 1024 };

char data_[max_length];

};

class Server {

public:

Server(asio::io_context& io_context, unsigned short port)

: acceptor_(io_context, tcp::endpoint(tcp::v4(), port)) {

do_accept();

}

private:

void do_accept() {

acceptor_.async_accept(

[this](asio::error_code ec, tcp::socket socket) {

if (!ec) {

std::make_shared<Session>(std::move(socket))->start();

}

do_accept();

});

}

tcp::acceptor acceptor_;

};

int main() {

try {

unsigned short port = 8888;

unsigned int thread_count = std::max(2u, std::thread::hardware_concurrency());

asio::io_context io_context;

Server server(io_context, port);

std::cout << "[multi-thread] echo server listening on port " << port

<< " with " << thread_count << " threads\n";

// 多个线程共同驱动同一个 io_context —— 这是 Asio 常见的线程池模式

std::vector<std::thread> threads;

threads.reserve(thread_count);

for (unsigned int i = 0; i < thread_count; ++i) {

threads.emplace_back([&io_context] { io_context.run(); });

}

for (auto& t : threads) t.join();

} catch (std::exception& e) {

std::cerr << "Exception: " << e.what() << "\n";

}

return 0;

}

协程版本

和前两版相比的关键变化
1. 不再有“回调套回调”，每个连接的读写逻辑写成一个线性的协程函数 echo()，看起来就像同步代码，但 co_await 处会挂起，不阻塞线程
2. 用 co_spawn 把 echo() 这个协程“扔”给 executor 去跑，detached 表示不关心它的结果
3. 生命周期管理也简化了：不再需要 shared_from_this，socket 的所有权直接被协程帧持有，协程退出（正常结束或抛异常）时自动清理
4. 如果想要多线程版协程，只需要像 echo_server_2 一样多起几个线程调用io_context.run()，并把 co_spawn 的第一个参数换成 asio::make_strand(io_context)，以保证同一个连接的协程不会被并发调度

#include <asio.hpp>

#include <asio/co_spawn.hpp>

#include <asio/detached.hpp>

#include <asio/awaitable.hpp>

#include <asio/use_awaitable.hpp>

#include <iostream>

using asio::ip::tcp;

using asio::awaitable;

using asio::co_spawn;

using asio::detached;

using asio::use_awaitable;

namespace this_coro = asio::this_coro;

awaitable<void> echo(tcp::socket socket) {

char data[1024];

try {

for (;;) {

std::size_t n = co_await socket.async_read_some(

asio::buffer(data), use_awaitable);

co_await asio::async_write(

socket, asio::buffer(data, n), use_awaitable);

}

} catch (std::exception& e) {

// 对端关闭连接时 async_read_some 会抛异常，属于正常退出路径

std::cerr << "session closed: " << e.what() << "\n";

}

awaitable<void> listener(unsigned short port) {

auto executor = co_await this_coro::executor;

tcp::acceptor acceptor(executor, tcp::endpoint(tcp::v4(), port));

for (;;) {

tcp::socket socket = co_await acceptor.async_accept(use_awaitable);

// 每接受一个连接，就把 echo() 作为一个新协程 spawn 出去，互不阻塞

co_spawn(executor, echo(std::move(socket)), detached);

}

int main() {

try {

unsigned short port = 8888;

asio::io_context io_context;

co_spawn(io_context, listener(port), detached);

std::cout << "[coroutine] echo server listening on port " << port << "\n";

io_context.run(); // 这里演示单线程；改多线程见文件头注释第 4 点

} catch (std::exception& e) {

std::cerr << "Exception: " << e.what() << "\n";

}

return 0;

}

模板

模板类型参数的默认值

template <typename MutableBufferSequence,

ASIO_COMPLETION_TOKEN_FOR(void (asio::error_code,

std::size_t)) ReadToken = default_completion_token_t<executor_type>>

auto async_read_some(const MutableBufferSequence& buffers,

ReadToken&& token = default_completion_token_t<executor_type>())

-> decltype(

async_initiate<ReadToken,

void (asio::error_code, std::size_t)>(

declval<initiate_async_receive>(), token,

buffers, socket_base::message_flags(0)))

{

return async_initiate<ReadToken,

void (asio::error_code, std::size_t)>(

initiate_async_receive(this), token,

buffers, socket_base::message_flags(0));

}

ReadToken 这个模板参数如果调用者没显式指定，就用 default_completion_token_t<executor_type> 顶上
注意这里默认值依赖前一个模板参数不存在的东西
1. 实际上它依赖的是类内已知的 executor_type（socket 类自己的成员类型，不是模板参数），这在类模板的成员函数里是合法的

template <typename MutableBufferSequence,

ASIO_COMPLETION_TOKEN_FOR(void (asio::error_code, std::size_t))

ReadToken = default_completion_token_t<executor_type>>

auto async_read_some(...)

ASIO_COMPLETION_TOKEN_FOR 只是 typename 的马甲
1. sig 完全没被使用，纯粹是文档作用
2. 告诉阅读者"这个模板参数必须是能配合 void(error_code, size_t) 这个签名使用的 completion token"
3. 在启用 C++20 concepts 的构建里，ASIO 会把这个宏换成真正的 concept 约束（比如 completion_token_for<sig>），此时 sig 才真正被用上做编译期检查

#define ASIO_COMPLETION_TOKEN_FOR(sig) typename

#define ASIO_COMPLETION_TOKEN_FOR2(sig0, sig1) typename

#define ASIO_COMPLETION_TOKEN_FOR3(sig0, sig1, sig2) typename

1	template <typename MutableBufferSequence, typename ReadToken = ...>

别名模板（`alias template`）+ 继承取内嵌类型

这是标准的 type traits 手法，和 std::enable_if_t / std::remove_reference_t 一模一样
default_completion_token_impl<T> 是真正干活的模板
1. 它针对不同的 T（不同的执行器类型）做特化，内部定义一个 type
default_completion_token<T> 通过继承拿到这个 type
default_completion_token_t<T> 是 C++11 风格的 _t 别名模板，等价于 typename default_completion_token<T>::type
1. 省去调用方每次写 typename ... ::type 的麻烦
也就是说：
1. default_completion_token_t<executor_type>
2. 这行代码本质是在问："这个执行器类型对应的默认 completion token 是什么？
3. 具体答案取决于 default_completion_token_impl 针对该执行器的特化

template <typename T, typename = void>

struct default_completion_token_impl

{

typedef deferred_t type;

};

template <typename T>

struct default_completion_token

: detail::default_completion_token_impl<T>

{

};

template <typename T>

using default_completion_token_t = typename default_completion_token<T>::type;

函数参数默认值引用模板参数构造对象

ReadToken&& 是万能引用（forwarding reference）
1. 因为 ReadToken 是函数模板自己的类型参数（不是被 const/其他修饰过的具体类型）
2. 所以 ReadToken&& 在模板实参推导时既能绑定左值也能绑定右值

1 2	auto async_read_some(const MutableBufferSequence& buffers, ReadToken&& token = default_completion_token_t<executor_type>())

默认值 default_completion_token_t<executor_type>() 是
1. 构造了一个该类型的临时对象
2. 前面那行只是拿到"类型"，这里多了一对 ()，是在用这个类型的默认构造函数生成一个实际的值

尾置返回类型 + `decltype` + `SFINAE`

为什么不直接写返回类型，而要这么绕一圈？
1. 因为这个函数的真实返回类型是不确定的，它取决于 ReadToken 是什么
2. 如果 token 是一个回调 lambda，返回类型可能是 void
3. 如果 token 是 asio::use_future，返回类型是 std::future<size_t>
4. 如果 token 是 asio::use_awaitable，返回类型是一个可 co_await 的对象
这个返回类型只有靠调用 async_initiate<...>(...) 这个表达式，让编译器真正做一次类型推导才能知道
1. 所以用 decltype(表达式) 把返回类型"借用"过来，而不是自己手写

auto async_read_some(...)

-> decltype(

async_initiate<ReadToken,

void (asio::error_code, std::size_t)>(

declval<initiate_async_receive>(), token,

buffers, socket_base::message_flags(0)))

另外这种写法还有一个副作用：SFINAE
1. 如果对某个 ReadToken 类型，async_initiate<...>(...) 这个表达式根本不合法（编译不通过），那么整个函数在重载决议阶段会被"悄悄丢弃"而不是报错，让编译器去尝试其他重载/特化
2. 这是模板元编程里常见的"用表达式合法性做开关"的技巧

`declval<T>()`

这里是 asio 自己实现的一份
可以在不需要真正构造对象的前提下，在 decltype 这种"只做类型推导、不生成代码"的上下文里，假装拿到一个 T 类型的右值
它只能出现在 decltype/sizeof 等不求值语境（unevaluated context）里，因为它其实没有函数体，只有声明
用它是因为
1. initiate_async_receive 这个类型可能没有默认构造函数，或者构造代价不明确
2. 用 declval 就不用真的造一个对象出来，只是为了让 decltype 能推出 async_initiate(...) 调用后的返回类型

1	declval<initiate_async_receive>()

`Completion Token`实现原理

核心思想是

同一个异步函数，通过 completion token 这个参数，决定它到底"以什么方式完成"（回调 / future / 协程 / 阻塞等待...），而不用给每种方式写一个重载
具体拆开看三个角色：

`default_completion_token_impl<T>`

给执行器绑定默认的 token 类型
1. 不同的执行器（executor_type）可能有不同的"默认异步风格"
2. 比如：
3. 普通的 io_context::executor_type 默认 token 通常是 deferred_t 或类似的东西
4. 如果这个执行器被 asio::use_awaitable 之类的东西"打了标记"（关联了一个默认 completion token），那 default_completion_token_impl 的对应特化会返回 use_awaitable_t，这样你在协程里写 co_await socket.async_read_some(buf) 就不用每次手写 asio::use_awaitable 了
这本质是编译期的"策略选择表"：T -> 该T对应的默认token类型，通过模板特化实现，没有运行时开销

ReadToken token参数

这次调用具体用什么方式完成
调用者传进来的可以是
1. 一个回调函数/lambda：void(error_code, size_t)
2. asio::use_future
3. asio::use_awaitable
4. asio::yield_context（Stackful coroutine）
5. 甚至自定义的 token
这些类型五花八门，但函数体完全不用关心它们的差异，因为这些差异全部被丢给了下一步

`async_initiate<ReadToken, Signature>(initiate_obj, token, args...)`

真正的分发中枢
1. 这是整个机制的核心

template <ASIO_COMPLETION_SIGNATURE... Signatures,

typename CompletionToken, typename Initiation, typename... Args>

inline typename enable_if_t<

!enable_if_t<

detail::are_completion_signatures<Signatures...>::value,

detail::async_result_has_initiate_memfn<

CompletionToken, Signatures...>>::value,

async_result<decay_t<CompletionToken>, Signatures...>

>::return_type

async_initiate(Initiation&& initiation, CompletionToken&& token, Args&&... args)

{

async_completion<CompletionToken, Signatures...> completion(token);

static_cast<Initiation&&>(initiation)(

static_cast<

typename async_result<decay_t<CompletionToken>,

Signatures...>::completion_handler_type&&>(

completion.completion_handler),

static_cast<Args&&>(args)...);

return completion.result.get();

}

它做两件事：
通过 async_result<decay_t<ReadToken>, Signature> 这个 trait，决定
1. 这次调用该返回什么类型给用户（void？future<size_t>？协程 awaitable？）
2. 需要不需要把 token 转换成一个真正的、符合 Signature 的回调（比如把 use_future 转换成一个内部回调，这个回调负责 set_value/set_exception 到 promise 上）
调用 initiate_obj（这里是 initiate_async_receive(this)）
1. 把转换后的真正回调、以及 buffers、flags 等参数传给它，由它去调用最底层平台相关的异步 I/O 实现
initiate_async_receive 之所以被封装成一个函数对象（initiation object）而不是直接内联调用底层函数，是因为
1. async_result 的某些特化（比如协程、future）需要"延后"或者"包一层"调用时机
2. 比如协程场景要先把 awaitable 的状态机搭好，再决定什么时候真正触发底层 I/O；直接写成普通函数调用没法插入这种中间层

整体串联

用户调用

socket.async_read_some(buf, asio::use_future)

│

▼

ReadToken = asio::use_future_t<>（模板实参推导得到，不用默认值）

│

▼

async_initiate<use_future_t<>, void(error_code,size_t)>(

initiate_async_receive(this), token, buf, 0)

│

▼

查 async_result<use_future_t<>, void(error_code,size_t)> 特化

│

├─ 构造一个 std::promise + 一个内部完成回调

├─ 调用 initiate_async_receive(this)(内部回调, buf, 0)

│ └─ 真正发起底层异步读操作

└─ 返回 promise.get_future() ← 这就是 decltype 推出来的返回类型

同步 I/O：服务端
code
同步 I/O：客户端
code
异步I/O：服务端
code
异步I/O：客户端
code
定时器
code
线程池
code
strand
概述
最直觉的解法
strand的思路
code：不用 strand
code：无脑加锁
code：strand
strand<Executor>
strand的原理
示例：TCP 连接管理器
协程封装
C++20 awaitable + co_spawn
绑定 strand 的协程
Stackful 协程：asio::spawn
awaitable和spawn
echo server
单线程
多线程
协程版本
模板
模板类型参数的默认值
别名模板（alias template）+ 继承取内嵌类型
函数参数默认值引用模板参数构造对象
尾置返回类型 + decltype + SFINAE
declval<T>()
Completion Token实现原理
核心思想是
default_completion_token_impl<T>
ReadToken token参数
async_initiate<ReadToken, Signature>(initiate_obj, token, args...)
整体串联

声明：本文为原创文章，版权归Aet所有，欢迎分享本文，转载请保留出处！

Bingliaolong 关注：0 粉丝：0

Everything will be better.

你可能也喜欢

热评文章

* 暂无

2026 年 8 月
一	二	三	四	五	六	日
				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

ASIO：学习一

同步 I/O：服务端

code

同步 I/O：客户端

code

异步I/O：服务端

code

异步I/O：客户端

code

定时器

code

线程池

code

strand

概述

最直觉的解法

strand的思路

code：不用 strand

code：无脑加锁

code：strand

strand<Executor>

strand的原理

示例：TCP 连接管理器

协程封装

C++20 awaitable + co_spawn

绑定 strand 的协程

Stackful 协程：asio::spawn

awaitable和spawn

echo server

单线程

多线程

协程版本

模板

模板类型参数的默认值

别名模板（alias template）+ 继承取内嵌类型

函数参数默认值引用模板参数构造对象

尾置返回类型 + decltype + SFINAE

declval<T>()

Completion Token实现原理

核心思想是

default_completion_token_impl<T>

ReadToken token参数

async_initiate<ReadToken, Signature>(initiate_obj, token, args...)

整体串联

你可能也喜欢

热评文章

发表评论

标签云集

同步 `I/O`：服务端

`code`

同步 `I/O`：客户端

`code`

异步`I/O`：服务端

`code`

异步`I/O`：客户端

`code`

`code`

`code`

`strand`

`strand`的思路

`code`：不用 `strand`

`code`：无脑加锁

`code：strand`

`strand<Executor>`

`strand`的原理

示例：`TCP` 连接管理器

`C++20 awaitable` + `co_spawn`

绑定 `strand` 的协程

`Stackful` 协程：`asio::spawn`

`awaitable`和`spawn`

`echo server`

别名模板（`alias template`）+ 继承取内嵌类型

尾置返回类型 + `decltype` + `SFINAE`

`declval<T>()`

`Completion Token`实现原理

`default_completion_token_impl<T>`

`async_initiate<ReadToken, Signature>(initiate_obj, token, args...)`