Tokio 在 Rust 中的应用：基础介绍 – wiki基地

作者： /

深入探索 Tokio 在 Rust 中的应用：基础介绍

随着现代软件系统变得越来越复杂，处理大量的并发连接、高吞吐量的 I/O 操作（如网络请求、文件读写）成为了常态。在传统的同步编程模型中，当一个任务需要等待一个 I/O 操作完成时，整个线程会被阻塞，无法执行其他任务。这在高并发场景下会导致资源的极大浪费和性能瓶颈。

为了解决这个问题，异步编程应运而生。异步编程允许程序在等待一个耗时操作（如网络请求）完成时，切换去执行其他准备就绪的任务，从而极大地提高了单个线程的利用率和系统的整体吞吐量。

Rust 语言通过内置的 async/await 语法提供了对异步编程的原生支持。然而，async fn 返回的是一个 Future，Future 本身只是一个描述了异步操作的“待完成”状态的结构，它并不会自己运行。要让 Future 真正执行并取得结果，需要一个“执行器”（Executor）或者说“异步运行时”（Asynchronous Runtime）来驱动它不断向前执行，直到完成。

在 Rust 的异步生态系统中，Tokio 无疑是最流行、功能最强大的异步运行时之一。它提供了一整套构建可靠、高性能、异步应用的工具和组件，包括异步 I/O、定时器、任务管理、异步同步原语等。

本文将带你深入探索 Tokio 的世界，从最基础的概念开始，理解 Tokio 如何协同 Rust 的 async/await 工作，并通过具体示例展示如何在 Rust 中使用 Tokio 构建基本的异步应用。

1. Rust 的 Async/Await 简述：未来的承诺 (Future)

在深入 Tokio 之前，我们首先快速回顾一下 Rust 的 async/await。

当你在 Rust 中定义一个异步函数时，你会使用 async fn 关键字：

rust
async fn my_async_task() -> String {
// 模拟一个异步操作，例如网络请求或文件读取
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
println!("Task finished after 1 second.");
"Result from async task".to_string()
}

调用 my_async_task() 并不会立即执行函数体内的代码，而是会返回一个实现了 Future trait 的类型。Future 代表了一个将来某个时候会产生一个值的计算。它有一个核心方法：poll。运行时会反复调用 poll 方法来检查 Future 是否已经准备好向前执行，或者是否已经完成并产生结果。

await 关键字则用在 Future 上，例如 some_future.await。当代码执行到 .await 时，如果 some_future 还没有完成，当前的异步任务就会暂停执行，将控制权交还给运行时。运行时就可以去执行其他准备好的任务。一旦 some_future 完成（或者取得了新的进展），运行时会得到通知，然后在合适的时机恢复之前暂停的任务，从 .await 的位置继续执行。

重要的是要理解，async/await 语法只是帮助我们编写看起来像同步代码的异步逻辑，它定义了异步任务的状态转换和暂停/恢复点（即 .await 的位置）。但谁来调用 Future 的 poll 方法？谁来管理暂停和恢复的任务？这就是异步运行时（Runtime）的工作，Tokio 就是一个典型的异步运行时。

2. Tokio 简介与核心概念：异步应用的引擎

Tokio 是一个事件驱动的、非阻塞的 I/O 平台，用于使用 Rust 编写异步应用程序。它提供了一系列构建异步应用所需的核心组件：

  1. 调度器 (Scheduler / Executor): 负责接收 Future 任务，并在合适的时机调用它们的 poll 方法，驱动它们向前执行。当一个任务在 .await 处暂停时，调度器会记住它的状态，并在 I/O 事件就绪或定时器到期时恢复它。
  2. 反应器 (Reactor): 负责与操作系统底层进行交互，监听 I/O 事件（如 TCP 连接建立、数据可读/可写、文件句柄上的事件）。当 I/O 事件发生时，反应器会通知调度器，调度器再唤醒等待该事件的任务。Tokio 在 Linux 上通常使用 epoll，在 macOS/BSD 上使用 kqueue，在 Windows 上使用 IOCP 来实现高效的反应器。
  3. 定时器 (Timer): 负责处理基于时间的事件，如延迟执行或周期性执行任务。
  4. 异步 I/O (Async I/O): 提供了异步版本的标准库 I/O 类型，如 TcpStream, TcpListener, File 等。这些类型与 Tokio 的反应器集成，使得 I/O 操作不再阻塞线程。
  5. 异步同步原语 (Async Synchronization Primitives): 提供了适用于异步上下文的同步工具，如 Mutex, Semaphore, Channel 等，用于在不同的异步任务之间安全地共享数据和通信。

简单来说，Tokio 就像是一个精密的协同操作系统，它接管了你的 async fn 任务，并在一个或多个 OS 线程上高效地调度它们，同时处理底层的 I/O 事件和时间事件，让你能够以非阻塞的方式编写高并发代码。

2.1 Tokio Runtime：启动异步世界

Tokio 的核心是 tokio::runtime::Runtime。它包含了调度器、反应器等所有组件。启动一个 Tokio 运行时，你的异步代码才能真正跑起来。

最常见和便捷的方式是使用 #[tokio::main] 属性宏。它可以自动创建一个 Tokio 运行时，并在其中执行你的 async fn main 函数：

“`rust

[tokio::main]

async fn main() {
println!(“Hello from inside Tokio runtime!”);
// 这里可以调用其他 async 函数或者使用 Tokio 的异步 API
my_async_task().await;
println!(“Main function finished.”);
}

async fn my_async_task() {
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
println!(“Async task completed.”);
}
“`

#[tokio::main] 实际上是一个语法糖，它等价于手动创建和管理一个运行时：

“`rust
fn main() {
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
println!(“Hello from inside manually created Tokio runtime!”);
my_async_task().await;
println!(“Main function finished.”);
});
}

async fn my_async_task() {
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
println!(“Async task completed.”);
}
“`

手动创建 Runtime 通常用于更复杂的场景，例如在同步代码中启动异步部分，或者对运行时进行更精细的配置（如线程数量）。对于大多数应用来说，#[tokio::main] 已经足够方便。

Tokio Runtime 支持两种主要的调度模式：

  • current_thread: 在调用 block_on 的当前 OS 线程上运行所有异步任务。适合客户端应用或只需要少量并发 I/O 的场景。
  • multi_thread (默认): 创建一个线程池，将异步任务分发到这些线程上并行执行。这是服务器应用和需要最大化 CPU 利用率的场景的首选。它能够充分利用多核 CPU 的能力。

#[tokio::main] 默认创建的是 multi_thread 运行时（除非你指定 #[tokio::main(flavor = "current_thread")]）。

3. 构建你的第一个 Tokio 应用

让我们从一个简单的例子开始，展示如何设置项目和编写基本的异步代码。

3.1 项目设置

首先，创建一个新的 Rust 项目：

bash
cargo new tokio_basic_app
cd tokio_basic_app

然后，在 Cargo.toml 文件中添加 Tokio 依赖。为了方便使用所有常用的 Tokio 功能（异步 I/O, time, sync, net, fs 等），通常会启用 full feature：

toml
[dependencies]
tokio = { version = "1", features = ["full"] }

如果你只需要特定的功能，可以按需选择 feature，例如：

toml
[dependencies]
tokio = { version = "1", features = ["rt-multi-thread", "net", "time"] }

启用 full feature 对于初学者来说是最简单的。

3.2 一个简单的异步程序

修改 src/main.rs 文件：

“`rust
use tokio::time::{sleep, Duration};

[tokio::main]

async fn main() {
println!(“主任务开始”);

// 调用一个异步函数并等待它完成
let result1 = perform_async_operation(1).await;
println!("第一个异步操作完成，结果: {}", result1);

// 调用另一个异步函数并等待它完成
let result2 = perform_async_operation(2).await;
println!("第二个异步操作完成，结果: {}", result2);

println!("主任务结束");

}

async fn perform_async_operation(id: i32) -> String {
println!(“异步操作 {} 开始…”, id);
// 模拟一个耗时操作，例如等待网络响应
sleep(Duration::from_secs(2)).await;
println!(“异步操作 {} 完成.”, id);
format!(“Data from operation {}”, id)
}
“`

运行这个程序：cargo run

输出会是这样：

主任务开始
异步操作 1 开始...
异步操作 1 完成.
第一个异步操作完成，结果: Data from operation 1
异步操作 2 开始...
异步操作 2 完成.
第二个异步操作完成，结果: Data from operation 2
主任务结束

尽管 perform_async_operation 内部有 sleep().await 等待，但在 main 函数中我们使用 await 依次调用了它们。这意味着第二个操作会在第一个操作完全完成后才开始。这看起来是同步的，但关键在于 sleep().await 期间，主线程并没有被阻塞，Tokio 运行时可以去执行其他任务（尽管在这个例子中没有其他任务）。

4. 异步 I/O：Tokio 的核心优势

异步 I/O 是 Tokio 最重要的能力之一。Tokio 提供了 tokio::io, tokio::fs, tokio::net 等模块，提供了非阻塞的 I/O 操作。

4.1 异步文件读写 (tokio::fs)

标准库的 std::fs 函数是阻塞的。在异步应用中直接使用它们会阻塞整个线程，导致性能下降。Tokio 提供了 tokio::fs 模块，提供了异步版本的文件操作。

“`rust
use tokio::fs::File;
use tokio::io::AsyncReadExt; // 需要这个 trait 来使用 read_to_string

[tokio::main]

async fn main() -> tokio::io::Result<()> {
println!(“开始异步文件读取…”);

let file_path = "example.txt";

// 创建一个示例文件（同步操作，但在实际应用中文件通常已存在）
// std::fs::write(file_path, "Hello, Tokio Async FS!").expect("无法创建文件");

// 尝试异步读取文件
match File::open(file_path).await {
    Ok(mut file) => {
        let mut contents = String::new();
        // 异步读取文件内容到字符串
        file.read_to_string(&mut contents).await?;
        println!("文件 {} 内容:\n{}", file_path, contents);
    }
    Err(e) => {
        eprintln!("无法打开或读取文件 {}: {}", file_path, e);
        // 在实际应用中，你可能需要更复杂的错误处理
    }
}


println!("异步文件读取结束.");

Ok(())

}
“`

你需要先手动创建一个名为 example.txt 的文件，例如内容为 Hello, Tokio Async FS!。运行程序，它会异步读取文件内容并打印。

4.2 异步网络：构建一个简单的 TCP 服务器 (tokio::net)

网络 I/O 是异步编程最常见的应用场景。Tokio 的 tokio::net 模块提供了异步的 TCP 和 UDP 支持。让我们构建一个简单的 TCP echo 服务器。

“`rust
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};

[tokio::main]

async fn main() -> Result<(), Box> {
// 绑定到本地地址 127.0.0.1:8080
let listener = TcpListener::bind(“127.0.0.1:8080”).await?;
println!(“服务器正在监听端口 8080…”);

// 持续接受新的连接
loop {
    // accept 方法会等待一个入站连接。
    // 当有新的连接到来时，它返回一个新的 TcpStream 和客户端地址
    let (mut socket, addr) = listener.accept().await?;

    println!("接受到连接来自: {}", addr);

    // 为每个连接创建一个新的异步任务 (task)。
    // 这样，服务器可以同时处理多个连接，而不会阻塞
    tokio::spawn(async move {
        // 在一个缓冲区中读取数据
        let mut buf = [0; 1024];

        // 循环处理连接中的数据
        loop {
            // 异步读取数据到缓冲区
            let n = match socket.read(&mut buf).await {
                // 读取 n 个字节成功
                Ok(n) if n == 0 => return, // 连接被对端关闭
                Ok(n) => n,
                // 读取发生错误
                Err(e) => {
                    eprintln!("读取 socket 错误: {}", e);
                    return; // 发生错误，关闭连接
                }
            };

            // 成功读取了 n 个字节，将这些数据异步写回给客户端
            if socket.write_all(&buf[0..n]).await.is_Err() {
                // 写入错误，也关闭连接
                eprintln!("写入 socket 错误");
                return;
            }
        }
    });
}

}
“`

运行这个服务器程序：cargo run

然后你可以使用 telnet 或 nc (netcat) 等工具连接到服务器进行测试：

bash
telnet 127.0.0.1 8080

输入一些文本并按回车，服务器应该会将你输入的文本原样返回。当你在 telnet 中输入 Ctrl + ] 然后输入 quit 退出时，服务器会检测到连接关闭。

这个例子展示了 Tokio 处理并发连接的能力：

  • TcpListener::bind().await：异步绑定端口。
  • listener.accept().await：异步等待客户端连接。当没有连接时，任务会暂停，不会阻塞主循环。
  • tokio::spawn(async move { ... })：对于每个新的连接，创建一个 新的 异步任务来处理。async move 确保闭包获取 socket 的所有权。这些任务由 Tokio 运行时并发调度。
  • socket.read().await 和 socket.write_all().await：在每个连接的任务中，使用异步方法读写数据。当读写操作等待时，当前连接的任务会暂停，但其他连接的任务可以继续执行。

这是 Tokio 高性能服务器的基础模式。

5. 任务管理与并发：tokio::spawn 和 tokio::join!

在 Tokio 中，Future 通常是通过 tokio::spawn 函数提交给运行时执行的。tokio::spawn 会创建一个新的异步任务，并在后台并发地运行一个 Future，它不会阻塞当前任务。

“`rust

[tokio::main]

async fn main() {
println!(“主任务开始”);

//  Spawn 两个并发任务
let task1 = tokio::spawn(async {
    println!("任务 1 开始");
    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
    println!("任务 1 完成");
    "结果来自任务 1" // 任务返回一个值
});

let task2 = tokio::spawn(async {
    println!("任务 2 开始");
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
    println!("任务 2 完成");
    "结果来自任务 2" // 任务返回一个值
});

// 主任务可以做其他事情，例如再启动一个任务
let task3 = tokio::spawn(async {
    println!("任务 3 (在主任务等待时运行)");
    "结果来自任务 3"
});

// 等待指定的任务完成并获取它们的返回值
// tokio::join! 宏可以等待多个 futures 或 join handles 并发完成
let (result1, result2) = tokio::join!(task1, task2);

println!("任务 1 返回: {:?}", result1); // 注意 JoinHandle 的结果是 Result<T, JoinError>
println!("任务 2 返回: {:?}", result2);

// 等待 task3 完成
let result3 = task3.await;
println!("任务 3 返回: {:?}", result3);

println!("主任务结束");

}
“`

运行这个例子，你会看到任务 1 和任务 2 几乎同时开始，但任务 2 会先于任务 1 完成，因为它的睡眠时间更短。任务 3 也会在 task1 和 task2 执行期间并发运行。tokio::join!(task1, task2) 会等待 task1 和 task2 都完成后，主任务才会继续执行。

tokio::spawn 返回一个 JoinHandle<T>，其中 T 是 spawned 任务的返回值类型。你可以 .await 这个 JoinHandle 来等待任务完成并获取结果。注意，JoinHandle::await 返回一个 Result<T, JoinError>，因为 spawned 任务可能会 panic。

tokio::join! 是一个方便的宏，可以并行地等待多个 Future 完成。它会同时 poll 多个 futures，并在它们都完成时返回各自的结果。

6. 定时器：管理时间事件 (tokio::time)

异步应用中经常需要处理时间相关的事件，比如延迟执行、设置超时或者周期性执行。Tokio 的 tokio::time 模块提供了这些功能。

最常用的就是 tokio::time::sleep，它是 std::thread::sleep 的异步版本。std::thread::sleep 会阻塞当前的 OS 线程，而 tokio::time::sleep 只会暂停当前的异步任务，释放线程去执行其他任务。

“`rust
use tokio::time::{sleep, Duration, Instant};

[tokio::main]

async fn main() {
let start = Instant::now();
println!(“主任务开始，时间: {:?}”, start);

// 异步睡眠 2 秒
println!("准备睡眠 2 秒...");
sleep(Duration::from_secs(2)).await;
println!("睡眠 2 秒结束，总耗时: {:?}", start.elapsed());

// 异步睡眠 1 秒
println!("准备睡眠 1 秒...");
sleep(Duration::from_secs(1)).await;
println!("睡眠 1 秒结束，总耗时: {:?}", start.elapsed());

println!("主任务结束，总耗时: {:?}", start.elapsed());

}
“`

运行这个程序，你会看到总耗时大约是 3 秒，这符合两次睡眠时间之和。但在每次 sleep().await 期间，Tokio 运行时是空闲的，可以执行其他任务。

tokio::time 还提供了 timeout (给一个操作设置超时时间) 和 interval (周期性触发) 等功能，它们在构建更复杂的异步应用时非常有用。

7. 异步同步原语：协调并发任务 (tokio::sync)

在并发的异步任务之间共享可变状态时，需要使用同步原语来避免数据竞争。标准库的 std::sync 中的原语（如 Mutex）是为 OS 线程设计的，它们的 lock() 方法是阻塞的。如果在异步任务中调用 std::sync::Mutex::lock()，即使只阻塞很短的时间，也会暂停整个线程上的所有异步任务，这违背了异步编程的初衷。

Tokio 提供了适用于异步上下文的同步原语 tokio::sync。这些原语的锁定/等待方法是 async 的，它们在等待时会暂停当前异步任务，而不是阻塞线程。

最常用的是 tokio::sync::Mutex：

“`rust
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::time::{sleep, Duration};

[tokio::main]

async fn main() {
// 使用 Arc> 在多个任务间共享可变数据
let shared_data = Arc::new(Mutex::new(0));

let mut handles = vec![];

for i in 0..5 {
    let data = Arc::clone(&shared_data); // 克隆 Arc 智能指针
    let handle = tokio::spawn(async move {
        println!("任务 {} 尝试获取锁...", i);
        let mut data = data.lock().await; // 异步获取锁

        println!("任务 {} 已获取锁，当前值: {}", i, *data);
        *data += 1; // 修改共享数据

        sleep(Duration::from_millis(100)).await; // 模拟持有锁期间的异步操作

        println!("任务 {} 修改后值: {}，释放锁.", i, *data);
        // 锁在 data 变量离开作用域时自动释放 (MutexGuard 的 Drop 实现)
    });
    handles.push(handle);
}

// 等待所有任务完成
for handle in handles {
    handle.await.expect("任务执行失败");
}

// 获取最终结果 (需要再次获取锁)
let final_data = shared_data.lock().await;
println!("所有任务完成，最终共享数据的值: {}", *final_data);

}
“`

运行这个程序，你会看到每个任务都会尝试获取锁。data.lock().await 会等待直到锁可用。当一个任务持有锁时，其他任务到达 lock().await 处会暂停，直到锁被释放。这种方式确保了在并发访问共享数据时的安全性，同时不会阻塞整个 OS 线程。

Tokio 还提供了其他异步同步原语，例如：

  • tokio::sync::RwLock: 读写锁。
  • tokio::sync::mpsc: 多生产者、单消费者异步通道，用于任务间传递消息。
  • tokio::sync::oneshot: 单次发送的异步通道。
  • tokio::sync::Semaphore: 信号量。

这些原语是构建复杂异步并发应用的基石。

8. 总结与进阶方向

通过上面的介绍和示例，你应该对 Tokio 的基本应用有了初步的了解：

  • 异步编程的必要性： 解决 I/O 密集型任务的性能瓶颈。
  • Rust 的 Async/Await： 定义 Future，描述异步操作。
  • Tokio 运行时： 驱动 Future 执行，管理 I/O 事件和时间事件。
  • #[tokio::main]： 方便地启动 Tokio 运行时。
  • tokio::spawn： 创建并发执行的异步任务。
  • tokio::join!： 并发等待多个任务完成。
  • 异步 I/O (tokio::fs, tokio::net)： 非阻塞地进行文件和网络操作。
  • 定时器 (tokio::time)： 处理异步时间事件。
  • 异步同步原语 (tokio::sync)： 安全地在异步任务间共享数据和通信。

Tokio 凭借其高性能、可靠性和丰富的生态系统，成为了 Rust 异步编程的首选。它被广泛应用于网络服务（如 HTTP 服务器、代理）、数据库驱动、命令行工具、嵌入式系统等需要高效处理并发 I/O 的场景。

这仅仅是 Tokio 的基础介绍。要构建更健壮、更复杂的异步应用，还需要学习更多进阶概念：

  • 错误处理： 如何在异步代码中优雅地处理错误，结合 ? 运算符和 Result。
  • 任务取消： 异步任务可能会被取消，理解取消的传播和清理资源。
  • tokio::select!： 同时等待多个异步操作，并在其中一个完成时继续执行，用于实现超时、race 条件等。
  • 流 (Streams)： 处理一系列异步产生的值，常用于处理连接中的连续数据块。
  • 更高级的网络服务： 使用基于 Tokio 的框架，如 Hyper (HTTP), Axum (Web 框架), Tonic (gRPC) 等。
  • 自定义 Executor/Reactor： 在特定场景下手动构建和配置 Runtime。

继续学习和实践是掌握 Tokio 的关键。查阅 Tokio 官方文档（tokio.rs）是深入理解其工作原理和高级用法的最佳途径。

异步编程在开始时可能会有些挑战，但一旦掌握了 Tokio 的核心概念和使用模式，你就能利用 Rust 的安全性和性能优势，构建出高效、可靠、可伸缩的异步应用。祝你在探索 Tokio 的旅程中一切顺利！

发表评论

您的邮箱地址不会被公开。 必填项已用 * 标注

滚动至顶部