Rust Concurrency Programming Cheatsheet

use std::thread;
use std::time::Duration;

fn main() {
	// 创建多个线程
    let mut handles = vec![];
	for id in 0..5 {
    	let handle = thread::spawn(|| {
        	println!("线程开始执行");
        	// 模拟耗时操作
        	thread::sleep(Duration::from_millis(500));
        	println!("线程执行完毕");
        	// 返回值会被join()获取
        	42+i
    	});
		handles.push(handle);
	}
    
    // 主线程继续执行
    println!("主线程继续执行其他工作");
    // 等待线程完成并获取返回值
	for handle in handles {
        let result = handle.join().unwrap();
    	println!("线程返回值: {}", result);
    }
}

use tokio::time::{sleep, Duration};
#[tokio::main]
async fn main() {
    // 创建一个异步任务
    let handle = tokio::spawn(async {
        println!("异步任务开始执行");
        // 模拟异步操作
        sleep(Duration::from_secs(1)).await;
        println!("异步任务完成");
		// 返回值会被await获取
        "完成"
    });
    // 等待异步任务完成并获取返回值
    let result = handle.await.unwrap();
    println!("异步任务返回值: {}", result);
}

use std::process::Command;
use std::io;

fn main() -> io::Result<()> {
    // 简单的进程创建并等待完成
    let status = Command::new("ls")
        .arg("-l")
        .status()?;
    println!("进程退出状态: {}", status);
    
    // 捕获进程输出
    let output = Command::new("echo")
        .arg("Hello from another process!")
        .output()?;
        
    if output.status.success() {
        let stdout = String::from_utf8_lossy(&output.stdout);
        println!("进程输出: {}", stdout);
    }
    
    // 创建进程但不等待完成
    let mut child = Command::new("sleep")
        .arg("2")
        .spawn()?;
        
    println!("子进程已启动，PID: {}", child.id());
    
    // 可以选择等待完成
    let status = child.wait()?;
    println!("子进程已完成，退出状态: {}", status);
    
    // 使用管道与进程通信
    let mut child = Command::new("cat")
        .stdin(std::process::Stdio::piped())
        .stdout(std::process::Stdio::piped())
        .spawn()?;
        
    // 向进程写入数据
    if let Some(mut stdin) = child.stdin.take() {
        use std::io::Write;
        stdin.write_all(b"Hello from Rust!")?;
        // 关闭stdin，否则cat会一直等待输入
        drop(stdin);
    }
    
    // 读取进程输出
    let output = child.wait_with_output()?;
    println!("cat输出: {}", String::from_utf8_lossy(&output.stdout));
    
    Ok(())
}

use std::sync::Arc;
use std::thread;
fn main() {
    let data = Arc::new(vec![1, 2, 3]);
    let mut handles = vec![];
    
    for i in 0..3 {
        let data_clone = Arc::clone(&data);
        let handle = thread::spawn(move || {
            println!("Thread {}: {:?}", i, data_clone);
        });
        handles.push(handle);
    }
    for handle in handles {
        handle.join().unwrap();
    }
}

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    // 创建一个共享计数器
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];
    
    // 创建10个线程，每个线程递增计数器10次
    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            for _ in 0..10 {
                // 获取锁
                let mut count = counter.lock().unwrap();
                // 修改共享数据
                *count += 1;
                // 锁在作用域结束时自动释放
            }
        });
        handles.push(handle);
    }
    
    // 等待所有线程完成
    for handle in handles {
        handle.join().unwrap();
    }
    
    // 输出最终结果
    println!("计数器最终值: {}", *counter.lock().unwrap());
}

use std::sync::{Arc, RwLock};
use std::thread;
use std::time::Duration;

struct Database {
    data: Vec,
}

fn main() {
    // 创建共享数据库
    let db = Arc::new(RwLock::new(Database {
        data: vec!["初始数据".to_string()],
    }));
    
    // 创建一个写入线程
    let db_writer = Arc::clone(&db);
    let writer = thread::spawn(move || {
        for i in 0..5 {
            // 获取写锁
            let mut database = db_writer.write().unwrap();
            // 修改数据
            database.data.push(format!("新数据 {}", i));
            println!("写入: 新数据 {}", i);
            // 写锁在作用域结束时释放
            drop(database); // 显式释放锁，让读取者有机会获取锁
            thread::sleep(Duration::from_millis(100));
        }
    });
    
    // 创建多个读取线程
    let mut readers = vec![];
    for id in 0..3 {
        let db_reader = Arc::clone(&db);
        let reader = thread::spawn(move || {
            for _ in 0..8 {
                // 获取读锁 - 多个读取者可以同时持有读锁
                let database = db_reader.read().unwrap();
                println!("读取者 {}: 数据量 = {}", id, database.data.len());
                // 读锁在作用域结束时释放
                thread::sleep(Duration::from_millis(50));
            }
        });
        readers.push(reader);
    }
    
    // 等待所有线程完成
    writer.join().unwrap();
    for reader in readers {
        reader.join().unwrap();
    }
}

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 id in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = tokio::spawn(async move {
            // 异步获取锁
            let mut count = counter.lock().await;
            *count += 1;
            println!("任务 {} 将计数器增加到 {}", id, *count);
            // 锁在作用域结束时释放
        });
        handles.push(handle);
    }
    
    // 等待所有任务完成
    for handle in handles {
        handle.await.unwrap();
    }
    
    // 输出最终结果
    println!("计数器最终值: {}", *counter.lock().await);
}

use tokio::sync::Semaphore;
use std::sync::Arc;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    // 创建信号量，最多允许3个并发任务
    let semaphore = Arc::new(Semaphore::new(3));
    let mut handles = vec![];
    
    // 创建8个任务，但最多只有3个能同时执行
    for id in 0..8 {
        let semaphore = Arc::clone(&semaphore);
        let handle = tokio::spawn(async move {
            // 获取许可
            let permit = semaphore.acquire().await.unwrap();
            println!("任务 {} 获得许可，开始执行", id);
            
            // 模拟工作
            sleep(Duration::from_millis(500)).await;
            
            println!("任务 {} 完成执行", id);
            // 许可在permit被drop时自动释放
            drop(permit);
        });
        handles.push(handle);
    }
    
    // 等待所有任务完成
    for handle in handles {
        handle.await.unwrap();
    }
}

use std::sync::{Arc, Barrier};
use std::thread;
use std::time::Duration;

fn main() {
    // 创建屏障，等待3个线程
    let barrier = Arc::new(Barrier::new(3));
    let mut handles = vec![];
    
    for id in 0..3 {
        let barrier = Arc::clone(&barrier);
        let handle = thread::spawn(move || {
            println!("线程 {} 开始执行第一阶段", id);
            thread::sleep(Duration::from_millis(100 * id)); // 模拟不同的工作时间
            println!("线程 {} 完成第一阶段", id);
            
            // 等待所有线程完成第一阶段
            let wait_result = barrier.wait();
            
            // 只有一个线程的wait()会返回BarrierWaitResult::Leader
            if wait_result.is_leader() {
                println!("所有线程完成第一阶段，继续执行");
            }
            
            println!("线程 {} 开始执行第二阶段", id);
            // 继续执行第二阶段...
        });
        handles.push(handle);
    }
    
    // 等待所有线程完成
    for handle in handles {
        handle.join().unwrap();
    }
}

use std::sync::Once;
use std::thread;

// 全局静态变量
static INIT: Once = Once::new();
static mut CONFIG: Option = None;

fn get_config() -> &'static str {
    // 确保初始化代码只执行一次
    INIT.call_once(|| {
        // 这段代码只会执行一次
        println!("正在初始化配置...");
        // 模拟耗时的初始化过程
        thread::sleep(std::time::Duration::from_millis(100));
        
        // 安全地初始化静态变量
        unsafe {
            CONFIG = Some(String::from("数据库配置信息"));
        }
        println!("配置初始化完成");
    });
    
    // 安全地访问已初始化的静态变量
    unsafe {
        CONFIG.as_ref().unwrap()
    }
}

fn main() {
    // 创建多个线程，都尝试初始化配置
    let mut handles = vec![];
    
    for id in 0..5 {
        let handle = thread::spawn(move || {
            println!("线程 {} 请求配置", id);
            let config = get_config();
            println!("线程 {} 获得配置: {}", id, config);
        });
        handles.push(handle);
    }
    
    // 等待所有线程完成
    for handle in handles {
        handle.join().unwrap();
    }
}

use parking_lot::{Mutex, RwLock};
use std::sync::Arc;
use std::thread;
use std::time::Duration;

fn main() {
    // parking_lot的Mutex不返回Result，API更简洁
    let counter = Arc::new(Mutex::new(0));
    
    // 创建线程递增计数器
    let counter_clone = Arc::clone(&counter);
    let handle = thread::spawn(move || {
        let mut count = counter_clone.lock(); // 不需要unwrap()
        *count += 1;
    });
    
    // parking_lot的RwLock也更简洁
    let data = Arc::new(RwLock::new(vec![1, 2, 3]));
    
    // 读取线程
    let data_clone = Arc::clone(&data);
    let read_handle = thread::spawn(move || {
        let values = data_clone.read(); // 不需要unwrap()
        println!("读取数据: {:?}", *values);
    });
    
    // 写入线程
    let data_clone = Arc::clone(&data);
    let write_handle = thread::spawn(move || {
        let mut values = data_clone.write(); // 不需要unwrap()
        values.push(4);
    });
    
    // 等待所有线程完成
    handle.join().unwrap();
    read_handle.join().unwrap();
    write_handle.join().unwrap();
    
    // 输出最终结果
    println!("计数器最终值: {}", *counter.lock());
    println!("数据最终值: {:?}", *data.read());
    
    // parking_lot还支持尝试加锁，不会阻塞
    if let Some(mut count) = counter.try_lock() {
        *count += 1;
        println!("成功获取锁并递增: {}", *count);
    } else {
        println!("锁被占用，无法获取");
    }
    
    // parking_lot支持带超时的锁
    let timeout = Duration::from_millis(100);
    if let Some(mut count) = counter.try_lock_for(timeout) {
        *count += 1;
        println!("在超时前成功获取锁: {}", *count);
    } else {
        println!("等待超时，无法获取锁");
    }
}

use std::sync::atomic::*;

// 整数类型
AtomicBool, AtomicI8, AtomicI16, AtomicI32, AtomicI64, AtomicIsize
AtomicU8, AtomicU16, AtomicU32, AtomicU64, AtomicUsize

// 指针类型
AtomicPtr

// Mutex - 有锁
let counter = Arc::new(Mutex::new(0u64));
let mut guard = counter.lock().unwrap(); // 获取锁，可能阻塞
*guard += 1; // 修改数据
// guard 被 drop 时释放锁

// AtomicU64 - 无锁
let counter = Arc::new(AtomicU64::new(0));
counter.fetch_add(1, Ordering::Relaxed); // 直接原子操作，不会阻塞等锁

// 原始代码
let mut x = 0;
let mut y = 0;
x = 1;           // 操作A
y = 2;           // 操作B
println!("{}", x);

// 编译器/CPU 可能重排为：
let mut x = 0;
let mut y = 0;
y = 2;           // 操作B 先执行
x = 1;           // 操作A 后执行  
println!("{}", x);
// 单线程下结果一样，但多线程下可能出问题

use std::sync::atomic::Ordering;

Ordering::Relaxed    // 只保证原子性，没有同步或顺序约束
Ordering::Acquire    // 获取屏障：阻止后续读写重排到此操作之前
Ordering::Release    // 释放屏障：阻止之前读写重排到此操作之后  
Ordering::AcqRel     // 同时具有获取和释放语义
Ordering::SeqCst     // 顺序一致性：全局统一的操作顺序

// 生产者线程
fn producer() {
    DATA.store(42, Ordering::Relaxed);      // 操作1：设置数据
    READY.store(true, Ordering::Release);   // 操作2：Release屏障
}
// Release 确保：
// 1. DATA.store(42, ...) 必须先执行完
// 2. READY.store(true, Release) 之前的操作不能重排到这之后
// 结果：数据总是在标志位之前被设置

// 错误示例：使用 Relaxed
fn consumer_wrong() {
    if READY.load(Ordering::Relaxed) {  // 操作1
        let value = DATA.load(Ordering::Relaxed);  // 操作2
        println!("Data: {}", value);
    }
}

// 正确示例：使用 Acquire  
fn consumer_correct() {
    if READY.load(Ordering::Acquire) {  // 操作1：Acquire屏障
        let value = DATA.load(Ordering::Relaxed);  // 操作2：不能重排到操作1之前
        println!("Data: {}", value);
    }
}

// 如果没有Acquire，CPU 可能的重排序：
// 1. let value = DATA.load(...) // 先读数据
// 2. if READY.load(...)         // 后检查就绪标志
// 结果：可能读到未初始化的数据！

// 决策树
fn choose_ordering() {
    // 1. 问：这个操作需要与其他操作同步吗？
    if needs_synchronization() {
        // 2. 问：是否需要全局一致的顺序？
        if needs_global_order() {
            // 使用 SeqCst
        } else {
            // 3. 问：是发布数据还是消费数据？
            if publishing_data() {
                // 使用 Release
            } else if consuming_data() {
                // 使用 Acquire  
            } else {
                // 既发布又消费，使用 AcqRel
            }
        }
    } else {
        // 只需要原子性，使用 Relaxed
        // 不需要同步：不关心与其他操作的顺序关系
    }
}

use std::sync::atomic::{AtomicU64, Ordering};

let atomic = AtomicU64::new(0);

// 读取
let value = atomic.load(Ordering::Relaxed);

// 存储
atomic.store(42, Ordering::Relaxed);

// 原子加法
let old_value = atomic.fetch_add(10, Ordering::Relaxed);

// 原子最大值更新
atomic.fetch_max(100, Ordering::Relaxed);

// 比较并交换
// atomic.compare_exchange(期望值, 新值, 成功时的内存顺序, 失败时的内存顺序)
// 如果当前值等于期望值，就替换为新值，否则不做任何操作。
// 多个线程可能同时尝试修改同一个值，我们需要确保只有一个线程成功
let result = atomic.compare_exchange(42, 99, Ordering::Relaxed, Ordering::Relaxed);

// 实现一个原子计数器，只在值小于100时递增
fn increment_if_less_than_100(counter: &AtomicU64) -> bool {
    loop {
        let current = counter.load(Ordering::Relaxed);
        if current >= 100 {
            return false; // 已经达到上限
        }
        // 尝试将 current 替换为 current + 1
        match counter.compare_exchange(current, current + 1, Ordering::Relaxed, Ordering::Relaxed) {
            Ok(_) => return true,  // 成功递增
            Err(_) => {
                // 失败说明值被其他线程改了，重试
                continue;
            }
        }
    }
}

use crossbeam_channel as channel;
use std::thread;
use std::time::Duration;

fn crossbeam_example() {
    // 1. 无界 channel
    let (tx, rx) = channel::unbounded::();
    
    // 2. 有界 channel  
    let (tx_bounded, rx_bounded) = channel::bounded::(10);
    
    // 3. 零容量 channel（同步交换）
    let (tx_sync, rx_sync) = channel::bounded::(0);
    
    // 发送端
    thread::spawn(move || {
        tx.send(42).unwrap();
        
        // 非阻塞发送
        match tx_bounded.try_send(100) {
            Ok(_) => println!("Sent successfully"),
            Err(channel::TrySendError::Full(_)) => println!("Channel full"),
            Err(channel::TrySendError::Disconnected(_)) => println!("Disconnected"),
        }
        
        // 带超时的发送
        if let Err(_) = tx_bounded.send_timeout(200, Duration::from_millis(100)) {
            println!("Send timeout");
        }
    });
    
    // 接收端 - 多种接收方式
    thread::spawn(move || {
        // 1. 阻塞接收
        let value = rx.recv().unwrap();
        
        // 2. 非阻塞接收
        match rx_bounded.try_recv() {
            Ok(val) => println!("Received: {}", val),
            Err(channel::TryRecvError::Empty) => println!("Channel empty"),
            Err(channel::TryRecvError::Disconnected) => println!("Disconnected"),
        }
        
        // 3. 带超时的接收
        match rx_bounded.recv_timeout(Duration::from_millis(100)) {
            Ok(val) => println!("Received: {}", val),
            Err(channel::RecvTimeoutError::Timeout) => println!("Receive timeout"),
            Err(channel::RecvTimeoutError::Disconnected) => println!("Disconnected"),
        }
        
        // 4. 选择操作（类似 Go 的 select）
        channel::select! {
            recv(rx) -> msg => {
                if let Ok(val) = msg {
                    println!("Got from rx: {}", val);
                }
            },
            recv(rx_bounded) -> msg => {
                if let Ok(val) = msg {
                    println!("Got from rx_bounded: {}", val);
                }
            },
            default => {
                println!("No message available");
            },
        }
    });
}

use crossbeam_channel as channel;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = channel::bounded::(10);
    
    // 创建多个生产者
    for producer_id in 0..3 {
        let tx_clone = tx.clone();
        
        thread::spawn(move || {
            for msg_num in 0..5 {
                let message = format!("生产者{}-消息{}", producer_id, msg_num);
                tx_clone.send(message).unwrap();
                thread::sleep(Duration::from_millis(100));
            }
            println!("生产者{} 完成发送", producer_id);
        });
    }
    
    // 创建多个消费者 (关键：rx.clone() 在 crossbeam 中是可以的！)
    for consumer_id in 0..2 {
        let rx_clone = rx.clone();
        
        thread::spawn(move || {
            while let Ok(message) = rx_clone.recv() {
                println!("消费者{} 收到: {}", consumer_id, message);
                thread::sleep(Duration::from_millis(200)); // 模拟处理时间
            }
            println!("消费者{} 退出", consumer_id);
        });
    }
    
    // 丢弃原始的 tx 和 rx，让消费者知道没有更多的消息了
    drop(tx);
    drop(rx);
    
    // 等待所有线程完成
    thread::sleep(Duration::from_secs(3));
}

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    let (tx, rx) = mpsc::channel::();
    
    // 用循环创建多个生产者
    for producer_id in 0..5 {
        let tx_clone = tx.clone();
        
        thread::spawn(move || {
            for msg_num in 0..3 {
                let message = format!("生产者{}-消息{}", producer_id, msg_num);
                tx_clone.send(message).unwrap();
                thread::sleep(Duration::from_millis(100 * (producer_id + 1)));
            }
            println!("生产者{} 完成发送", producer_id);
        });
    }
    
    // 丢弃原始 tx，这样当所有生产者完成后 channel 会自动关闭
    drop(tx);
    
    // 单个消费者接收所有消息
    while let Ok(message) = rx.recv() {
        println!("消费者收到: {}", message);
    }
    
    println!("所有消息处理完毕");
}

// 典型的 broadcast 用法：关闭信号
use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (shutdown_tx, _) = broadcast::channel(1);
    
    // 启动多个任务
    for i in 0..3 {
        let mut shutdown_rx = shutdown_tx.subscribe();
        tokio::spawn(async move {
            tokio::select! {
                _ = some_work() => println!("任务{}正常完成", i),
                _ = shutdown_rx.recv() => println!("任务{}收到关闭信号", i),
            }
        });
    }
    
    // 发送关闭信号给所有任务
    shutdown_tx.send(()).unwrap();
}

async fn some_work() {
    tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
}