阶段四 · 并发编排与 IO

异步 IO 基础

一句话总结

异步 IO 由两个 trait 支撑:AsyncRead 和 AsyncWrite,它们是同步 std::io::Read/Write 的异步版本--方法返回 Future,读写未就绪时让出而非阻塞.
配合 AsyncReadExt/AsyncWriteExt 提供的 read/write_all 等便捷方法,以及带缓冲的 BufReader,就能写出高效的异步读写.

和同步 IO 的平行关系

异步 IO 不是全新的东西,它和已有的同步 IO 是一一对应的.把握这个对应,学起来就快:

同步(std::io) 异步(tokio::io) 差异
Read trait AsyncRead trait 读方法返回 Future,需 .await
Write trait AsyncWrite trait 写方法返回 Future,需 .await
read() 阻塞线程 read().await 让出任务 未就绪时的行为不同
BufReader tokio::io::BufReader 用法几乎一致

异步 IO 就是给同步 IO 的每个读写动作"加了一个 .await".
骨架完全相同--打开,读,写,缓冲;唯一变化是:遇到数据没就绪,同步版冻结线程,异步版让出任务.已有的 IO 知识几乎可以平移.

两个核心 trait 与扩展方法

AsyncRead / AsyncWrite 是底层 trait(定义了 poll_read 之类的原始方法,和第 03 篇的 poll 同源),但日常基本不直接碰它们.真正常用的是它们的扩展 trait:AsyncReadExt 和 AsyncWriteExt,提供了好用的高层方法.

use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::fs::File;

async fn copy_demo() -> std::io::Result<()> {
    let mut f = File::open("in.txt").await?;
    let mut buf = Vec::new();
    f.read_to_end(&mut buf).await?;       // 读全部到 buf

    let mut out = File::create("out.txt").await?;
    out.write_all(&buf).await?;           // 写出全部
    out.flush().await?;                   // 刷新缓冲
    Ok(())
}

别忘了 use 扩展 trait

如果只 use tokio::fs::File 却没 use tokio::io::AsyncReadExt,会发现 read,write_all 这些方法"不存在".

这是 Rust 的扩展 trait 机制:方法定义在 trait 里,必须把 trait 引入作用域才能调用其方法.这是异步 IO 新手最常见的"方法找不到"困惑.

常用读写方法

方法 来自 作用
read(&mut buf) AsyncReadExt 读一次,返回读到的字节数(可能少于缓冲)
read_to_end(&mut vec) AsyncReadExt 读到 EOF 为止
read_exact(&mut buf) AsyncReadExt 读满整个缓冲,不够则报错
write(&buf) AsyncWriteExt 写一次,返回实际写入字节数
write_all(&buf) AsyncWriteExt 保证写完整个缓冲
flush() AsyncWriteExt 刷新底层缓冲

单次 read 返回的字节数可能小于给定的缓冲大小--网络数据是流式到达的,这是正常现象,不是错误.返回 0 才表示对端关闭(EOF).

需要精确读取定长数据时用 read_exact.这条规则和同步 IO 完全一致,Go 的 conn.Read 也是如此.

带缓冲读写:BufReader 与按行读取

频繁的小块读写会产生大量系统调用,低效.和同步世界一样,用缓冲包装器减少系统调用次数.BufReader 还提供按行读取:

use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::fs::File;

async fn read_lines() -> std::io::Result<()> {
    let file = File::open("log.txt").await?;
    let reader = BufReader::new(file);
    let mut lines = reader.lines();

    while let Some(line) = lines.next_line().await? {
        println!("行:{line}");
    }
    Ok(())
}

io::copy:流式转发的利器

把一个读端的数据全部搬到一个写端,不用手写循环--tokio::io::copy 一行搞定.它对实现代理,转发非常有用:

use tokio::io;

io::copy(&mut reader, &mut writer).await?;

承上启下

有了 AsyncReadExt/AsyncWriteExt 和缓冲读写,已经具备处理网络连接数据流的全部基础工具.下一篇就把它们和 TcpListener,spawn 组装起来,亲手写出一个能并发处理多连接的异步 TCP 服务器.

快速回顾

  • 平行对应:AsyncRead/AsyncWrite 是 std Read/Write 的异步版,本质是"读写动作加 .await".
  • 扩展 trait:常用方法来自 AsyncReadExt/AsyncWriteExt,必须 use 进作用域才能调用.
  • read 语义:单次读可能不满缓冲(流式到达属正常),返回 0 表示 EOF;定长用 read_exact.
  • 缓冲:BufReader/BufWriter 减少系统调用;lines() 支持按行读取.
  • io::copy:流式把读端数据全量转发到写端,代理/转发场景的利器.

动手练习

  1. 文件读写:用 tokio::fs::File + read_to_end 读一个文件,再 write_all 写到另一个文件.
  2. 复现报错:故意不 use AsyncReadExt,观察 read 方法"找不到"的报错,再加上 use 修复.
  3. 按行读取:用 BufReader::lines 逐行读取一个文本文件并打印行号.
  4. 流式复制:用 tokio::io::copy 把一个文件的内容流式复制到另一个文件.
  5. 解释语义:为什么单次 read 返回的字节数可能小于缓冲大小?返回 0 又意味着什么?