🦀 第15课：async/await异步编程

Rust的异步编程模型基于Future特征和async/await语法。与线程不同，异步任务是轻量级的协作式并发，单线程就能处理数千个并发操作。这是构建高性能网络服务的核心工具。

进阶特性第15/25课

学习目标：理解Future和async/await、掌握tokio运行时、异步IO操作、并发任务组合

🔮 Future特征

Future是异步计算的核心抽象，表示一个可能尚未完成的值：

// Future特征定义（简化版）
// pub trait Future {
//     type Output;
//     fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll;
// }
// 
// pub enum Poll {
//     Ready(T),      // 值已就绪
//     Pending,       // 尚未就绪，稍后再poll
// }

// async函数返回一个Future
async fn say_hello() {
    println!("Hello, async Rust! 🦀");
}

// async函数可以返回值
async fn add(a: i32, b: i32) -> i32 {
    a + b
}

fn main() {
    // Future是惰性的——必须被poll才会执行
    let future = say_hello();
    
    // 使用block_on来执行future
    // 在实际项目中使用tokio::main，这里演示概念
    println!("异步函数尚未执行...");
    // 需要一个执行器来运行future
    // 实际项目中用 #[tokio::main] 标注main
}

⚡ tokio运行时

tokio是Rust最流行的异步运行时，提供事件循环、任务调度和异步IO：

// Cargo.toml:
// [dependencies]
// tokio = { version = "1", features = ["full"] }

// 使用tokio的完整示例
// #[tokio::main]
// async fn main() {
//     println!("Hello from tokio!");
// }

// 这里用同步代码演示概念
fn main() {
    println!("=== async/await概念演示 ===\n");
    
    // 概念1: async fn创建Future
    // async fn fetch_data(url: &str) -> Result {
    //     let response = reqwest::get(url).await?;  // .await暂停直到完成
    //     let body = response.text().await?;
    //     Ok(body)
    // }
    
    // 概念2: .await点
    // .await是异步函数中的"暂停点"
    // 当前Future返回Pending，让出执行权
    // 当IO就绪后，从await点恢复执行
    
    // 概念3: 并发vs并行
    // 并发(concurrent): 多个任务交替执行（单线程也可以）
    // 并行(parallel): 多个任务同时执行（需要多线程）
    
    println!("async fn → 返回Future");
    println!(".await → 暂停并等待Future完成");
    println!("tokio::spawn → 创建并发任务");
    println!("join! → 等待多个Future完成");
    println!("select! → 等待第一个完成的Future");
    println!("FuturesUnordered → 动态任务集合");
}

// 模拟异步操作的同步版本
mod mock_async {
    use std::thread;
    use std::time::Duration;
    
    pub fn delay_ms(ms: u64) {
        thread::sleep(Duration::from_millis(ms));
    }
    
    pub fn fetch_user(id: u32) -> String {
        delay_ms(100);  // 模拟网络延迟
        format!("User_{}", id)
    }
    
    pub fn fetch_posts(user: &str) -> Vec {
        delay_ms(150);
        vec![
            format!("{}的第1篇文章", user),
            format!("{}的第2篇文章", user),
        ]
    }
}

fn main_demo() {
    use mock_async::*;
    
    println!("模拟异步流程:");
    let user = fetch_user(42);
    println!("获取用户: {}", user);
    
    let posts = fetch_posts(&user);
    println!("获取文章: {:?}", posts);
}

🔄 并发任务组合

// 实际tokio代码模式（概念展示）

// 1. 串行执行（一个接一个）
// async fn sequential() {
//     let user = fetch_user(1).await;     // 等待完成
//     let posts = fetch_posts(&user).await; // 再等待
// }

// 2. 并发执行（同时开始）
// async fn concurrent() {
//     let (user, config) = tokio::join!(
//         fetch_user(1),
//         fetch_config(),
//     );  // 两个同时进行
// }

// 3. 竞争执行（取最快）
// async fn racing() {
//     tokio::select! {
//         result = fetch_from_cache() => {
//             println!("缓存命中: {}", result);
//         }
//         result = fetch_from_db() => {
//             println!("数据库查询: {}", result);
//         }
//     }
// }

// 4. 动态并发
// async fn many_tasks() {
//     let mut handles = vec![];
//     for i in 0..10 {
//         handles.push(tokio::spawn(async move {
//             fetch_user(i).await
//         }));
//     }
//     for handle in handles {
//         let result = handle.await.unwrap();
//         println!("{}", result);
//     }
// }

use std::thread;
use std::time::{Duration, Instant};

fn main() {
    // 模拟并发vs串行
    println!("⏱️ 性能对比:");
    
    // 串行
    let start = Instant::now();
    thread::sleep(Duration::from_millis(100));  // 模拟任务1
    thread::sleep(Duration::from_millis(150));  // 模拟任务2
    println!("串行耗时: {:?}", start.elapsed());  // ~250ms
    
    // 模拟并发（用线程模拟）
    let start = Instant::now();
    let h1 = thread::spawn(|| thread::sleep(Duration::from_millis(100)));
    let h2 = thread::spawn(|| thread::sleep(Duration::from_millis(150)));
    h1.join().unwrap();
    h2.join().unwrap();
    println!("并发耗时: {:?}", start.elapsed());  // ~150ms
    
    // 异步的优势：一个线程处理大量IO
    // 1000个网络请求 × 100ms/请求:
    //   串行: 100秒
    //   1000线程: 内存爆炸
    //   异步: ~100ms（单线程并发）
    println!("\n异步适合的场景: 大量IO密集型操作");
    println!("线程适合的场景: CPU密集型计算");
}

// 异步流（Stream）
// use tokio_stream;
// async fn stream_example() {
//     let mut stream = tokio_stream::iter(1..=5);
//     while let Some(value) = stream.next().await {
//         println!("收到: {}", value);
//     }
// }

=== async/await概念演示 === async fn → 返回Future .await → 暂停并等待Future完成 tokio::spawn → 创建并发任务 join! → 等待多个Future完成 select! → 等待第一个完成的Future FuturesUnordered → 动态任务集合 ⏱️ 性能对比: 串行耗时: 250.xxxms 并发耗时: 150.xxxms 异步适合的场景: 大量IO密集型操作线程适合的场景: CPU密集型计算

✅ 验证通过

🏗️ 综合实战：模拟异步HTTP服务

use std::thread;
use std::time::{Duration, Instant};
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

// 模拟异步HTTP请求
struct AsyncHttpClient {
    cache: Arc>>,
}

impl AsyncHttpClient {
    fn new() -> Self {
        let mut cache = HashMap::new();
        cache.insert("https://api.rust-lang.org/version".to_string(), "1.95.0".to_string());
        cache.insert("https://crates.io/top".to_string(), "serde, tokio, clap".to_string());
        AsyncHttpClient { cache: Arc::new(Mutex::new(cache)) }
    }
    
    fn get(&self, url: &str) -> String {
        // 模拟网络延迟
        thread::sleep(Duration::from_millis(50));
        self.cache.lock().unwrap()
            .get(url)
            .cloned()
            .unwrap_or_else(|| format!("404: {}", url))
    }
}

// 模拟异步任务调度
struct AsyncTask {
    name: String,
    duration_ms: u64,
}

impl AsyncTask {
    fn new(name: &str, duration_ms: u64) -> Self {
        AsyncTask { name: name.to_string(), duration_ms }
    }
    
    fn execute(&self) -> String {
        thread::sleep(Duration::from_millis(self.duration_ms));
        format!("✅ {} ({}ms)", self.name, self.duration_ms)
    }
}

fn main() {
    let client = AsyncHttpClient::new();
    
    // 串行请求
    println!("📡 串行请求:");
    let start = Instant::now();
    let v = client.get("https://api.rust-lang.org/version");
    let t = client.get("https://crates.io/top");
    println!("  版本: {}", v);
    println!("  热门: {}", t);
    println!("  耗时: {:?}", start.elapsed());
    
    // 并发请求
    println!("\n📡 并发请求:");
    let start = Instant::now();
    let cache = client.cache.clone();
    let h1 = thread::spawn(move || {
        thread::sleep(Duration::from_millis(50));
        cache.lock().unwrap().get("https://api.rust-lang.org/version").cloned().unwrap()
    });
    let cache2 = client.cache.clone();
    let h2 = thread::spawn(move || {
        thread::sleep(Duration::from_millis(50));
        cache2.lock().unwrap().get("https://crates.io/top").cloned().unwrap()
    });
    println!("  版本: {}", h1.join().unwrap());
    println!("  热门: {}", h2.join().unwrap());
    println!("  耗时: {:?}", start.elapsed());
    
    // 多任务并发
    println!("\n🚀 多任务并发:");
    let tasks: Vec = vec![
        AsyncTask::new("任务A", 100),
        AsyncTask::new("任务B", 150),
        AsyncTask::new("任务C", 80),
        AsyncTask::new("任务D", 120),
    ];
    
    let start = Instant::now();
    let handles: Vec<_> = tasks.into_iter().map(|t| {
        thread::spawn(move || t.execute())
    }).collect();
    
    for h in handles {
        println!("  {}", h.join().unwrap());
    }
    println!("  总耗时: {:?}", start.elapsed());
    println!("  (理论串行: 450ms, 实际并发: ~150ms)");
}

📡 串行请求: 版本: 1.95.0 热门: serde, tokio, clap 耗时: 100.xxxms 📡 并发请求: 版本: 1.95.0 热门: serde, tokio, clap 耗时: 50.xxxms 🚀 多任务并发: ✅ 任务C (80ms) ✅ 任务A (100ms) ✅ 任务D (120ms) ✅ 任务B (150ms) 总耗时: 150.xxxms (理论串行: 450ms, 实际并发: ~150ms)

✅ 验证通过

📝 练习

练习1：Tokio入门

创建一个tokio项目，编写3个异步函数，用join!并发执行。

练习2：异步定时器

使用tokio::time实现每隔1秒输出当前时间的异步循环。

练习3：异步文件读取

使用tokio::fs并发读取多个文件，统计总行数。

🏆 本课成就

✅ 理解Future和async/await的工作原理
✅ 掌握tokio运行时的基本使用
✅ 理解并发与并行的区别
✅ 学会join!/select!任务组合
✅ 理解异步编程的适用场景

🔒 下一课解锁：模块与包管理 —— 组织你的Rust项目

🔧 async/await完整示例

// 使用tokio的完整示例模式：
// 
// [dependencies]
// tokio = { version = "1", features = ["full"] }
// reqwest = { version = "0.12", features = ["json"] }
// serde = { version = "1", features = ["derive"] }
//
// #[tokio::main]
// async fn main() -> Result<(), Box> {
//     // 并发请求
//     let (r1, r2, r3) = tokio::join!(
//         reqwest::get("https://httpbin.org/get"),
//         reqwest::get("https://httpbin.org/ip"),
//         reqwest::get("https://httpbin.org/headers"),
//     );
//     println!("{:?}", r1?.text().await?);
//     Ok(())
// }

// 同步模拟演示
use std::thread;
use std::time::{Duration, Instant};

fn mock_async_task(name: &str, duration_ms: u64) -> String {
    thread::sleep(Duration::from_millis(duration_ms));
    format!("{}完成({}ms)", name, duration_ms)
}

fn main() {
    // 串行 vs 并发 vs 异步概念
    println!("=== 串行执行 ===");
    let start = Instant::now();
    let r1 = mock_async_task("任务A", 100);
    let r2 = mock_async_task("任务B", 150);
    let r3 = mock_async_task("任务C", 80);
    println!("{} {} {}", r1, r2, r3);
    println!("串行耗时: {:?}", start.elapsed());
    
    println!("
=== 并发执行 ===");
    let start = Instant::now();
    let h1 = thread::spawn(|| mock_async_task("任务A", 100));
    let h2 = thread::spawn(|| mock_async_task("任务B", 150));
    let h3 = thread::spawn(|| mock_async_task("任务C", 80));
    println!("{} {} {}", h1.join().unwrap(), h2.join().unwrap(), h3.join().unwrap());
    println!("并发耗时: {:?}", start.elapsed());
    
    println!("
=== 异步概念 ===");
    println!("async/await: 单线程即可实现并发效果");
    println!("适用: IO密集型(网络/文件/数据库)");
    println!("不适用: CPU密集型(计算/加密/压缩)");
    
    // select!概念
    println!("
=== select! 模式 ===");
    println!("select! 取最先完成的结果:");
    println!("  tokio::select! {{");
    println!("      result = task_a() => handle_a(result),");
    println!("      result = task_b() => handle_b(result),");
    println!("  }}");
}

=== 串行执行 === 任务A完成(100ms) 任务B完成(150ms) 任务C完成(80ms) 串行耗时: 330.xxxms === 并发执行 === 任务A完成(100ms) 任务B完成(150ms) 任务C完成(80ms) 并发耗时: 150.xxxms === 异步概念 === async/await: 单线程即可实现并发效果适用: IO密集型(网络/文件/数据库) 不适用: CPU密集型(计算/加密/压缩)

✅ 验证通过

← 上一课：并发编程 📖 目录下一课：模块与包管理 →