Rust 并发、异步编程实践经验实用指南

use std::thread;
use std::time::Duration;
fn main() {
    // 1. 创建可连接线程，执行闭包任务
    let handle = thread::spawn(|| {
        for i in 1..=5 {
            println!("子线程任务：{}", i);
            thread::sleep(Duration::from_millis(100)); // 模拟耗时操作
        }
        "子线程执行完成" // 线程返回值
    });
    // 主线程任务
    for i in 1..=3 {
        println!("主线程任务：{}", i);
        thread::sleep(Duration::from_millis(150));
    }
    // 等待子线程结束，获取返回值（若不调用 join，子线程可能被强制终止）
    let result = handle.join().unwrap();
    println!("子线程返回：{}", result);
}

use std::thread;
fn main() {
    let msg = String::from("Hello, Rust Thread!");
    // 使用 move 转移 msg 所有权到子线程
    let handle = thread::spawn(move || {
        println!("子线程捕获的消息：{}", msg);
    });
    handle.join().unwrap();
    // 此处 msg 所有权已转移，无法再使用
    // println!("{}", msg); // 编译错误：value borrowed here after move
}

use std::sync::mpsc;
use std::thread;
use std::time::Duration;
fn main() {
    // 创建通道：返回 (发送者 Sender, 接收者 Receiver)
    let (tx, rx) = mpsc::channel();
    // 创建多个生产者线程
    let tx1 = tx.clone(); // 克隆发送者（支持多生产者）
    thread::spawn(move || {
        let messages = vec![
            String::from("消息1：来自生产者1"),
            String::from("消息2：来自生产者1"),
            String::from("消息3：来自生产者1"),
        ];
        for msg in messages {
            tx1.send(msg).unwrap(); // 发送消息
            thread::sleep(Duration::from_millis(200));
        }
    });
    thread::spawn(move || {
        let messages = vec![
            String::from("消息1：来自生产者2"),
            String::from("消息2：来自生产者2"),
        ];
        for msg in messages {
            tx.send(msg).unwrap();
            thread::sleep(Duration::from_millis(300));
        }
    });
    // 主线程作为消费者，接收所有消息
    for received in rx {
        println!("主线程接收：{}", received);
    }
    // 当所有发送者被销毁，通道关闭，for 循环自动退出
}

use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
    // Arc 包裹 Mutex，实现跨线程共享可修改数据
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];
    // 创建 10 个线程，每个线程对计数器加 1
    for _ in 0..10 {
        let counter_clone = Arc::clone(&counter); // 克隆 Arc（原子操作，轻量）
        let handle = thread::spawn(move || {
            let mut num = counter_clone.lock().unwrap(); // 获取互斥锁，阻塞直到成功
            *num += 1; // 临界区：修改共享数据
            // 离开作用域，num 被销毁，自动释放锁
        });
        handles.push(handle);
    }
    // 等待所有线程执行完毕
    for handle in handles {
        handle.join().unwrap();
    }
    // 最终计数器值为 10（无数据竞争）
    println!("最终计数器值：{}", *counter.lock().unwrap());
}

use std::sync::{Arc, RwLock};
use std::thread;
use std::time::Duration;
fn main() {
    let data = Arc::new(RwLock::new(String::from("初始数据")));
    let mut handles = vec![];
    // 3 个读线程
    for i in 0..3 {
        let data_clone = Arc::clone(&data);
        handles.push(thread::spawn(move || {
            let read_data = data_clone.read().unwrap(); // 获取读锁
            println!("读线程 {} 读取数据：{}", i, read_data);
            thread::sleep(Duration::from_millis(100)); // 模拟读操作耗时
        }));
    }
    // 1 个写线程
    let data_clone = Arc::clone(&data);
    handles.push(thread::spawn(move || {
        let mut write_data = data_clone.write().unwrap(); // 获取写锁，阻塞直到所有读锁释放
        *write_data = String::from("修改后的数据");
        println!("写线程修改数据完成");
    }));
    for handle in handles {
        handle.join().unwrap();
    }
    // 最终读取修改后的数据
    println!("最终数据：{}", data.read().unwrap());
}

use std::rc::Rc;
use std::sync::Mutex;
fn main() {
    // Rc 未实现 Send，无法跨线程传递
    let rc = Rc::new(5);
    // thread::spawn(move || { println!("{}", rc); }); // 编译错误：Rc<i32> cannot be sent between threads safely
    // Arc 实现了 Send + Sync，可跨线程共享
    let arc = std::sync::Arc::new(5);
    thread::spawn(move || { println!("{}", arc); }); // 正常编译
    // Mutex<T> 实现 Send，若 T 实现 Send + Sync，则 Mutex<T> 实现 Sync
    let mutex = Mutex::new(5);
    thread::spawn(move || { println!("{}", mutex.lock().unwrap()); }); // 正常编译
}

// 简化版 Future 特质（标准库中定义更复杂，包含上下文和唤醒机制）
trait SimpleFuture {
    type Output;
    // poll 方法：尝试推进任务执行，返回任务状态
    fn poll(&mut self) -> std::task::Poll<Self::Output>;
}
// 实现一个简单的 Future：延迟返回值
struct DelayFuture {
    delay: Duration,
    start_time: Option<Instant>,
}
impl DelayFuture {
    fn new(delay: Duration) -> Self {
        DelayFuture {
            delay,
            start_time: None,
        }
    }
}
impl SimpleFuture for DelayFuture {
    type Output = String;
    fn poll(&mut self) -> std::task::Poll<Self::Output> {
        let start = self.start_time.get_or_insert(Instant::now());
        if start.elapsed() >= self.delay {
            // 任务完成，返回结果
            std::task::Poll::Ready(String::from("延迟任务完成"))
        } else {
            // 任务未完成，告知执行器后续唤醒后再次 poll
            std::task::Poll::Pending
        }
    }
}

// 需引入 tokio 作为异步执行器（Cargo.toml 添加：tokio = { version = "1.0", features = ["full"] }）
use tokio::time::{sleep, Duration};
// async 函数：返回一个实现了 Future 的匿名类型，Output 为 String
async fn async_task(name: &str, delay: Duration) -> String {
    sleep(delay).await; // 等待延迟，非阻塞，让出执行权给其他任务
    format!("{} 执行完成", name)
}
#[tokio::main] // tokio 提供的属性宏，初始化异步执行器
async fn main() {
    // 并发执行多个异步任务，无需手动创建线程
    let task1 = async_task("任务1", Duration::from_millis(200));
    let task2 = async_task("任务2", Duration::from_millis(100));
    let task3 = async_task("任务3", Duration::from_millis(300));
    // 使用 join! 宏等待多个任务完成，返回结果元组
    let (result1, result2, result3) = tokio::join!(task1, task2, task3);
    println!("{}", result1); // 任务1 执行完成
    println!("{}", result2); // 任务2 执行完成
    println!("{}", result3); // 任务3 执行完成
}

// Cargo.toml 添加：async-std = { version = "1.0", features = ["attributes"] }
use async_std::task;
use std::time::Duration;
async fn async_task(name: &str, delay: Duration) -> String {
    task::sleep(delay).await;
    format!("{} 执行完成", name)
}
#[async_std::main] // async-std 执行器初始化
async fn main() {
    let task1 = async_task("任务1", Duration::from_millis(200));
    let task2 = async_task("任务2", Duration::from_millis(100));
    let (result1, result2) = futures::join!(task1, task2); // futures 库提供的 join 宏
    println!("{}", result1);
    println!("{}", result2);
}

use tokio::time::{timeout, Duration};
use tokio::task::{self, AbortHandle};
async fn long_running_task() -> String {
    sleep(Duration::from_secs(5)).await;
    String::from("长时间任务完成")
}
#[tokio::main]
async fn main() {
    // 1. 任务取消
    let (abort_handle, abort_registration) = AbortHandle::new_pair();
    let task = task::spawn(async move {
        task::abort_on_drop(abort_registration); // 绑定取消注册
        long_running_task().await
    });
    sleep(Duration::from_secs(2)).await;
    abort_handle.abort(); // 取消任务
    match task.await {
        Ok(_) => println!("任务正常完成"),
        Err(e) => println!("任务被取消：{}", e), // 输出：任务被取消：task aborted
    }
    // 2. 任务超时
    let result = timeout(Duration::from_secs(2), long_running_task()).await;
    match result {
        Ok(msg) => println!("{}", msg),
        Err(_) => println!("任务超时"), // 输出：任务超时
    }
}

use tokio::stream::{self, StreamExt};
use tokio::time::Duration;
#[tokio::main]
async fn main() {
    // 创建一个异步流：产生 1-5 的整数，每个间隔 100ms
    let mut stream = stream::iter(1..=5)
        .then(|num| async move {
            sleep(Duration::from_millis(100)).await;
            num * 2 // 每个元素乘以 2
        });
    // 遍历流中的值
    while let Some(val) = stream.next().await {
        println!("流元素：{}", val); // 依次输出 2、4、6、8、10
    }
}

use tokio::task;
use std::fs;
// 阻塞函数：同步读取文件
fn read_file_sync(path: &str) -> String {
    fs::read_to_string(path).unwrap()
}
#[tokio::main]
async fn main() {
    // 将阻塞代码放入阻塞线程池执行，不影响异步执行器
    let file_content = task::spawn_blocking(|| {
        read_file_sync("test.txt")
    }).await.unwrap();
    println!("文件内容长度：{}", file_content.len());
}

use std::thread;
use tokio::runtime::Runtime;
use tokio::time::{sleep, Duration};
async fn async_task() -> String {
    sleep(Duration::from_millis(100)).await;
    String::from("线程中的异步任务完成")
}
fn main() {
    // 在新线程中初始化 Tokio 运行时，运行异步任务
    thread::spawn(|| {
        let rt = Runtime::new().unwrap();
        let result = rt.block_on(async_task()); // 阻塞线程，等待异步任务完成
        println!("{}", result);
    }).join().unwrap();
}

use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
    let lock1 = Arc::new(Mutex::new(1));
    let lock2 = Arc::new(Mutex::new(2));
    let l1 = Arc::clone(&lock1);
    let l2 = Arc::clone(&lock2);
    thread::spawn(move || {
        let _g1 = l1.lock().unwrap();
        thread::sleep(Duration::from_millis(100));
        let _g2 = l2.lock().unwrap(); // 等待线程2释放 lock2，死锁
    });
    let l1 = Arc::clone(&lock1);
    let l2 = Arc::clone(&lock2);
    thread::spawn(move || {
        let _g2 = l2.lock().unwrap();
        thread::sleep(Duration::from_millis(100));
        let _g1 = l1.lock().unwrap(); // 等待线程1释放 lock1，死锁
    });
    thread::sleep(Duration::from_secs(1));
}

use tokio::task;
use tokio::time::yield_now;
async fn cpu_intensive_task() {
    for i in 0..1_000_000 {
        if i % 100_000 == 0 {
            yield_now().await; // 每计算 10 万次，让出执行权
        }
        // 模拟计算
        let _ = i * i;
    }
    println!("计算任务完成");
}
#[tokio::main]
async fn main() {
    let task1 = cpu_intensive_task();
    let task2 = async {
        task::sleep(Duration::from_millis(100)).await;
        println!("异步任务2完成");
    };
    tokio::join!(task1, task2); // 任务2不会被任务1阻塞
}

use tokio::sync::Mutex;
use std::sync::Arc;
#[tokio::main]
async fn main() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];
    for _ in 0..5 {
        let c = Arc::clone(&counter);
        handles.push(tokio::spawn(async move {
            let mut num = c.lock().await; // 异步锁，非阻塞
            *num += 1;
        }));
    }
    for h in handles {
        h.await.unwrap();
    }
    println!("计数器值：{}", *counter.lock().await); // 输出：5
}