异步编程
异步编程是一种编程范式，允许程序在等待 I/O 操作完成时执行其他任务，而不是阻塞当前线程。Rust 的异步编程模型基于 Future 特质，结合 async/await 语法，提供了高效、安全的异步编程体验。在本章中，我们将探索 Rust 的异步编程模型和工具。
同步 vs 异步
在深入异步编程之前，让我们先了解同步和异步编程的区别：
同步编程
在同步编程模型中，当程序执行 I/O 操作（如读取文件或网络请求）时，当前线程会阻塞，直到操作完成：
use std::fs::File;
use std::io::Read;fn read_file() -> std::io::Result<String> {let mut file = File::open("hello.txt")?; // 阻塞，直到文件打开let mut contents = String::new();file.read_to_string(&mut contents)?; // 阻塞，直到文件读取完成Ok(contents)
}fn main() {// 这个调用会阻塞主线程，直到文件读取完成let contents = read_file().unwrap();println!("{}", contents);
}
异步编程
在异步编程模型中，I/O 操作不会阻塞当前线程，而是返回一个表示未来完成操作的值（在 Rust 中是 Future）：
use tokio::fs::File;
use tokio::io::AsyncReadExt;async fn read_file() -> std::io::Result<String> {let mut file = File::open("hello.txt").await?; // 不阻塞，返回 Futurelet mut contents = String::new();file.read_to_string(&mut contents).await?; // 不阻塞，返回 FutureOk(contents)
}#[tokio::main]
async fn main() {// 这个调用不会阻塞主线程let contents = read_file().await.unwrap();println!("{}", contents);
}
Future 特质
Rust 的异步编程基于 Future 特质，它表示一个可能尚未完成的值：
pub trait Future {type Output;fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}pub enum Poll<T> {Ready(T),Pending,
}
Future 特质的核心是 poll 方法，它尝试解析 Future 的值。如果值已准备好，它返回 Poll::Ready(value)；如果值尚未准备好，它返回 Poll::Pending 并安排在值准备好时再次调用 poll。
手动实现 Future
虽然通常不需要手动实现 Future，但了解其工作原理很有帮助：
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};struct Delay {when: Instant,
}impl Future for Delay {type Output = ();fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {if Instant::now() >= self.when {println!("Future 已完成");Poll::Ready(())} else {// 安排在未来某个时刻再次调用 polllet waker = cx.waker().clone();let when = self.when;std::thread::spawn(move || {let now = Instant::now();if now < when {std::thread::sleep(when - now);}waker.wake();});println!("Future 尚未完成");Poll::Pending}}
}#[tokio::main]
async fn main() {let delay = Delay {when: Instant::now() + Duration::from_secs(2),};println!("等待 Future 完成...");delay.await;println!("Future 已完成，程序继续执行");
}
async/await 语法
Rust 提供了 async/await 语法，简化了异步编程：
async 函数
async 关键字用于定义返回 Future 的函数：
async fn say_hello() {println!("Hello");
}// 等价于：
fn say_hello() -> impl Future<Output = ()> {async {println!("Hello");}
}
await 表达式
.await 用于等待 Future 完成并获取其结果：
async fn get_user_id() -> u64 {// 模拟异步操作tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;42
}async fn get_user_name(id: u64) -> String {// 模拟异步操作tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;format!("User {}", id)
}#[tokio::main]
async fn main() {let id = get_user_id().await;let name = get_user_name(id).await;println!("{}", name);
}
async 块
async 也可以用于创建异步代码块：
#[tokio::main]
async fn main() {let future = async {let id = get_user_id().await;let name = get_user_name(id).await;name};let name = future.await;println!("{}", name);
}
异步运行时
Rust 的标准库提供了 Future 特质，但没有提供执行 Future 的运行时。为此，我们需要使用第三方库，如 Tokio、async-std 或 smol。
Tokio
Tokio 是 Rust 生态系统中最流行的异步运行时：
# Cargo.toml
[dependencies]
tokio = { version = "1", features = ["full"] }
use tokio::time::{sleep, Duration};async fn task_one() {println!("任务一：开始");sleep(Duration::from_millis(100)).await;println!("任务一：完成");
}async fn task_two() {println!("任务二：开始");sleep(Duration::from_millis(50)).await;println!("任务二：完成");
}#[tokio::main]
async fn main() {// 串行执行task_one().await;task_two().await;println!("---");// 并发执行tokio::join!(task_one(),task_two());
}
async-std
async-std 是另一个流行的异步运行时，API 设计与标准库类似：
# Cargo.toml
[dependencies]
async-std = { version = "1", features = ["attributes"] }
use async_std::task;
use std::time::Duration;async fn task_one() {println!("任务一：开始");task::sleep(Duration::from_millis(100)).await;println!("任务一：完成");
}async fn task_two() {println!("任务二：开始");task::sleep(Duration::from_millis(50)).await;println!("任务二：完成");
}#[async_std::main]
async fn main() {// 串行执行task_one().await;task_two().await;println!("---");// 并发执行let t1 = task::spawn(task_one());let t2 = task::spawn(task_two());t1.await;t2.await;
}
异步并发工具
并发执行多个 Future
tokio::join!
tokio::join! 宏并发执行多个 Future，等待所有 Future 完成：
use tokio::time::{sleep, Duration};#[tokio::main]
async fn main() {let (result1, result2) = tokio::join!(async {sleep(Duration::from_millis(100)).await;"结果一"},async {sleep(Duration::from_millis(50)).await;"结果二"});println!("{}, {}", result1, result2);
}
futures::join!
futures 库也提供了类似的 join! 宏：
use futures::join;
use async_std::task;
use std::time::Duration;#[async_std::main]
async fn main() {let (result1, result2) = join!(async {task::sleep(Duration::from_millis(100)).await;"结果一"},async {task::sleep(Duration::from_millis(50)).await;"结果二"});println!("{}, {}", result1, result2);
}
任务生成
tokio::spawn
tokio::spawn 在 Tokio 运行时中生成一个新任务：
use tokio::time::{sleep, Duration};#[tokio::main]
async fn main() {let handle = tokio::spawn(async {sleep(Duration::from_secs(1)).await;"Hello from task"});// 做其他工作...let result = handle.await.unwrap();println!("{}", result);
}
async_std::task::spawn
async_std::task::spawn 在 async-std 运行时中生成一个新任务：
use async_std::task;
use std::time::Duration;#[async_std::main]
async fn main() {let handle = task::spawn(async {task::sleep(Duration::from_secs(1)).await;"Hello from task"});// 做其他工作...let result = handle.await;println!("{}", result);
}
选择第一个完成的 Future
tokio::select!
tokio::select! 宏等待多个 Future 中的第一个完成：
use tokio::time::{sleep, Duration};#[tokio::main]
async fn main() {tokio::select! {_ = sleep(Duration::from_secs(1)) => {println!("1 秒过去了");}_ = sleep(Duration::from_secs(2)) => {println!("2 秒过去了");}}// 输出：1 秒过去了
}
futures::select!
futures 库也提供了类似的 select! 宏：
use futures::select;
use futures::future::{self, FutureExt};
use async_std::task;
use std::time::Duration;#[async_std::main]
async fn main() {let a = task::sleep(Duration::from_secs(1)).fuse();let b = task::sleep(Duration::from_secs(2)).fuse();select! {_ = a => println!("1 秒过去了"),_ = b => println!("2 秒过去了"),}// 输出：1 秒过去了
}
异步 I/O
文件操作
use tokio::fs::File;
use tokio::io::{self, AsyncReadExt, AsyncWriteExt};#[tokio::main]
async fn main() -> io::Result<()> {// 异步写入文件let mut file = File::create("hello.txt").await?;file.write_all(b"Hello, world!").await?;// 异步读取文件let mut file = File::open("hello.txt").await?;let mut contents = String::new();file.read_to_string(&mut contents).await?;println!("{}", contents);Ok(())
}
网络操作
use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {// 服务器let server = tokio::spawn(async {let listener = TcpListener::bind("127.0.0.1:8080").await.unwrap();let (mut socket, _) = listener.accept().await.unwrap();let mut buf = [0; 1024];let n = socket.read(&mut buf).await.unwrap();println!("服务器收到: {}", String::from_utf8_lossy(&buf[..n]));socket.write_all(b"Hello from server").await.unwrap();});// 客户端let client = tokio::spawn(async {let mut stream = TcpStream::connect("127.0.0.1:8080").await.unwrap();stream.write_all(b"Hello from client").await.unwrap();let mut buf = [0; 1024];let n = stream.read(&mut buf).await.unwrap();println!("客户端收到: {}", String::from_utf8_lossy(&buf[..n]));});// 等待两个任务完成tokio::try_join!(server, client)?;Ok(())
}
异步流（Stream）
Stream 特质类似于同步的 Iterator，但是异步产生值：
use futures::stream::{self, StreamExt};#[tokio::main]
async fn main() {// 创建一个简单的流let mut stream = stream::iter(vec![1, 2, 3, 4, 5]);// 使用 next 方法异步迭代流while let Some(value) = stream.next().await {println!("{}", value);}// 使用组合器方法处理流let sum = stream::iter(vec![1, 2, 3, 4, 5]).fold(0, |acc, x| async move { acc + x }).await;println!("Sum: {}", sum);
}
实际示例：处理 TCP 连接流
use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use futures::stream::StreamExt;#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {let listener = TcpListener::bind("127.0.0.1:8080").await?;let mut incoming = tokio_stream::wrappers::TcpListenerStream::new(listener);while let Some(socket_result) = incoming.next().await {let mut socket = socket_result?;// 为每个连接生成一个新任务tokio::spawn(async move {let mut buf = [0; 1024];loop {match socket.read(&mut buf).await {Ok(0) => return, // 连接关闭Ok(n) => {// 回显收到的数据if socket.write_all(&buf[..n]).await.is_err() {return;}}Err(_) => return,}}});}Ok(())
}
异步编程的挑战
错误处理
异步代码中的错误处理与同步代码类似，但需要注意 .await 的位置：
use tokio::fs::File;
use tokio::io::{self, AsyncReadExt};#[tokio::main]
async fn main() -> io::Result<()> {// 方法 1：使用 ? 运算符async fn read_file() -> io::Result<String> {let mut file = File::open("hello.txt").await?;let mut contents = String::new();file.read_to_string(&mut contents).await?;Ok(contents)}// 方法 2：使用 match 表达式async fn read_file_alt() -> io::Result<String> {let file_result = File::open("hello.txt").await;let mut file = match file_result {Ok(file) => file,Err(e) => return Err(e),};let mut contents = String::new();match file.read_to_string(&mut contents).await {Ok(_) => Ok(contents),Err(e) => Err(e),}}let contents = read_file().await?;println!("{}", contents);Ok(())
}
异步上下文限制
在 Rust 中，不能在同步函数中直接使用 .await：
// 错误：不能在同步函数中使用 .await
fn sync_function() {let future = async { "Hello" };let result = future.await; // 错误！
}
解决方案是使整个函数异步，或使用运行时的阻塞方法：
// 方法 1：使整个函数异步
async fn async_function() {let future = async { "Hello" };let result = future.await;println!("{}", result);
}// 方法 2：使用运行时的阻塞方法
fn sync_function() {let future = async { "Hello" };let result = tokio::runtime::Runtime::new().unwrap().block_on(future);println!("{}", result);
}
生命周期问题
异步函数中的引用可能导致复杂的生命周期问题：
// 错误：返回的 Future 包含对 name 的引用，但 name 在函数返回时已经离开作用域
async fn process(name: &str) -> String {format!("处理: {}", name)
}fn main() {let future;{let name = String::from("Alice");future = process(&name); // 错误：name 的生命周期不够长}// name 已经离开作用域，但 future 仍然持有对它的引用
}
解决方案是调整生命周期或使用所有权：
// 方法 1：调整生命周期
async fn process<'a>(name: &'a str) -> String {format!("处理: {}", name)
}// 方法 2：使用所有权
async fn process_owned(name: String) -> String {format!("处理: {}", name)
}fn main() {let name = String::from("Alice");let future = process_owned(name);// 现在 future 拥有 name 的所有权
}
异步编程最佳实践
1. 避免阻塞异步运行时
在异步函数中避免使用阻塞操作，如 std::thread::sleep 或同步 I/O：
// 不好的做法
async fn bad_practice() {// 这会阻塞整个异步运行时线程！std::thread::sleep(std::time::Duration::from_secs(1));
}// 好的做法
async fn good_practice() {tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
}
对于无法避免的阻塞操作，使用 spawn_blocking：
#[tokio::main]
async fn main() {// 在单独的线程中执行阻塞操作let result = tokio::task::spawn_blocking(|| {// 这里可以执行阻塞操作std::thread::sleep(std::time::Duration::from_secs(1));"完成"}).await.unwrap();println!("{}", result);
}
2. 适当使用并发
使用 join! 或 spawn 并发执行独立任务：
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {// 并发获取多个资源let (users, posts) = tokio::join!(fetch_users(),fetch_posts());// 处理结果for user in users? {println!("用户: {}", user);}for post in posts? {println!("文章: {}", post);}Ok(())
}async fn fetch_users() -> Result<Vec<String>, Box<dyn std::error::Error>> {// 模拟网络请求tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;Ok(vec![String::from("Alice"), String::from("Bob")])
}async fn fetch_posts() -> Result<Vec<String>, Box<dyn std::error::Error>> {// 模拟网络请求tokio::time::sleep(tokio::time::Duration::from_millis(150)).await;Ok(vec![String::from("Post 1"), String::from("Post 2")])
}
3. 使用超时
为异步操作设置超时，避免无限等待：
use tokio::time::{timeout, Duration};#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {// 设置 1 秒超时match timeout(Duration::from_secs(1), slow_operation()).await {Ok(result) => println!("操作完成: {}", result?),Err(_) => println!("操作超时"),}Ok(())
}async fn slow_operation() -> Result<String, Box<dyn std::error::Error>> {// 模拟慢操作tokio::time::sleep(Duration::from_secs(2)).await;Ok(String::from("操作结果"))
}
4. 使用适当的错误处理
在异步代码中使用 ? 运算符和 Result 类型：
use tokio::fs::File;
use tokio::io::{self, AsyncReadExt};#[tokio::main]
async fn main() -> io::Result<()> {let result = read_and_process_file("config.txt").await;match result {Ok(content) => println!("处理结果: {}", content),Err(e) => eprintln!("错误: {}", e),}Ok(())
}async fn read_and_process_file(path: &str) -> io::Result<String> {let mut file = File::open(path).await.map_err(|e| io::Error::new(e.kind(), format!("无法打开文件 {}: {}", path, e)))?;let mut content = String::new();file.read_to_string(&mut content).await.map_err(|e| io::Error::new(e.kind(), format!("无法读取文件 {}: {}", path, e)))?;// 处理内容Ok(format!("处理后的内容: {}", content))
}
5. 使用 Stream 处理异步数据流
对于需要处理异步数据流的场景，使用 Stream 特质：
use futures::stream::{self, StreamExt};
use tokio::time::{sleep, Duration};#[tokio::main]
async fn main() {// 创建一个异步数据流let mut stream = stream::unfold(0, |state| async move {if state < 5 {sleep(Duration::from_millis(100)).await;Some((state, state + 1))} else {None}});// 使用 for_each 并发处理流中的每个项stream.for_each_concurrent(2, |item| async move {println!("处理项: {}", item);sleep(Duration::from_millis(200)).await;println!("项 {} 处理完成", item);}).await;
}
练习题
-
创建一个异步函数，它接受一个文件路径，读取文件内容，并返回文件中的单词数量。使用 Tokio 运行时执行这个函数。
-
实现一个简单的异步 HTTP 客户端，它可以并发地从多个 URL 获取内容，并返回每个 URL 的内容长度。使用
reqwest库和tokio::join!宏。 -
创建一个异步 TCP 回显服务器，它接受客户端连接，并将收到的所有数据回显给客户端。使用 Tokio 的
TcpListener和TcpStream。 -
实现一个异步函数，它模拟一个可能失败的操作，并在失败时自动重试最多三次，每次重试之间有递增的延迟。使用 Tokio 的
sleep函数。 -
创建一个异步
_web前端开发设计_荆州百度推广_百度一下首页手机版)
