Rust: Rust 并发编程：线程、消息传递与共享状态

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

fn main() {
    // 创建一个共享文档内容（Arc<Mutex<String>>）
    let document = Arc::new(Mutex::new(String::from("# 协作文档\n\n")));

    // 创建一个通道，用于传输编辑请求
    let (tx, rx) = mpsc::channel::<String>();

    // 接收线程：持续处理编辑请求
    let doc_for_receiver = Arc::clone(&document);
    let receiver = thread::spawn(move || {
        for edit in rx {
            let mut doc = doc_for_receiver.lock().unwrap();
            doc.push_str(&edit);
            doc.push('\n');
            println!("[接收器] 已应用编辑");
        }
    });

    // 模拟编辑者线程
    let tx1 = tx.clone();
    thread::spawn(move || {
        tx1.send("- Alice: 添加了第一章内容".to_string()).unwrap();
    });

    let tx2 = tx.clone();
    thread::spawn(move || {
        tx2.send("- Bob: 修改了第二章标题".to_string()).unwrap();
    });

    // 主线程也发送一条
    tx.send("- Luna: 格式调整".to_string()).unwrap();

    // 等待所有编辑发送完成
    thread::sleep(Duration::from_millis(100));
    drop(tx);  // 关闭发送端

    receiver.join().unwrap();

    // 最终文档内容
    let final_doc = document.lock().unwrap();
    println!("\n=== 最终文档 ===");
    println!("{}", *final_doc);
}

// ============================================
// 演示：thread::spawn 创建线程、join 等待结束
// 模拟：厨房里多个厨师同时做不同的菜
// ============================================

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

fn main() {
    println!("=== 厨房开工 ===");

    // --- 1. 创建三个线程执行不同的任务 ---
    let chef1 = thread::spawn(|| {
        for i in 1..=3 {
            println!("[厨师A] 切菜中... 第{}刀", i);
            thread::sleep(Duration::from_millis(50));
        }
        "A 完成了切菜"
    });

    let chef2 = thread::spawn(|| {
        for i in 1..=3 {
            println!("[厨师B] 炒菜中... 第{}步", i);
            thread::sleep(Duration::from_millis(40));
        }
        "B 完成了炒菜"
    });

    // 主线程也在工作
    for i in 1..=3 {
        println!("[主厨] 装盘中... 第{}个", i);
        thread::sleep(Duration::from_millis(60));
    }

    // --- 2. join 等待所有线程结束并获取返回值 ---
    let result1 = chef1.join().unwrap();
    let result2 = chef2.join().unwrap();

    println!("\n=== 厨房收工 ===");
    println!("厨师A: {}", result1);
    println!("厨师B: {}", result2);
    println!("所有工作完成！");
}

=== 厨房开工 ===
[厨师A] 切菜中... 第1刀
[厨师B] 炒菜中... 第1步
[主厨] 装盘中... 第1个
[厨师A] 切菜中... 第2刀
[厨师B] 炒菜中... 第2步
[主厨] 装盘中... 第2个
[厨师A] 切菜中... 第3刀
[厨师B] 炒菜中... 第3步
[主厨] 装盘中... 第3个

=== 厨房收工 ===
厨师A: A 完成了切菜
厨师B: B 完成了炒菜
所有工作完成！

// ============================================
// 演示：mpsc::channel 多生产者单消费者消息传递
// 模拟：多个编辑者向文档服务器发送修改请求
// ============================================

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

#[derive(Debug)]
enum EditAction {
    Insert { user: String, text: String },
    Delete { user: String, line: u32 },
    Format { user: String, style: String },
}

fn main() {
    println!("=== 协作文档编辑器 ===");

    // 创建通道：Sender 可以克隆，Receiver 是唯一的
    let (tx, rx) = mpsc::channel::<EditAction>();

    // --- 1. 启动接收线程（文档服务器）---
    let receiver = thread::spawn(move || {
        for action in rx {
            match &action {
                EditAction::Insert { user, text } => {
                    println!("[服务器] {} 插入了: {}", user, text);
                }
                EditAction::Delete { user, line } => {
                    println!("[服务器] {} 删除了第{}行", user, line);
                }
                EditAction::Format { user, style } => {
                    println!("[服务器] {} 应用了格式: {}", user, style);
                }
            }
            // 模拟处理时间
            thread::sleep(Duration::from_millis(20));
        }
        println!("[服务器] 通道关闭，停止接收");
    });

    // --- 2. 创建第一个编辑者线程（Alice）---
    let tx1 = tx.clone();
    let editor1 = thread::spawn(move || {
        tx1.send(EditAction::Insert {
            user: "Alice".to_string(),
            text: "第一章：Rust 简介".to_string(),
        }).unwrap();
        thread::sleep(Duration::from_millis(10));
        tx1.send(EditAction::Format {
            user: "Alice".to_string(),
            style: "标题加粗".to_string(),
        }).unwrap();
    });

    // --- 3. 创建第二个编辑者线程（Bob）---
    let tx2 = tx.clone();
    let editor2 = thread::spawn(move || {
        tx2.send(EditAction::Insert {
            user: "Bob".to_string(),
            text: "Rust 是一门系统编程语言".to_string(),
        }).unwrap();
        thread::sleep(Duration::from_millis(10));
        tx2.send(EditAction::Delete {
            user: "Bob".to_string(),
            line: 1,
        }).unwrap();
    });

    // --- 4. 主线程也发送一条消息 ---
    tx.send(EditAction::Insert {
        user: "系统".to_string(),
        text: "自动保存文档".to_string(),
    }).unwrap();

    // 等待编辑者线程结束
    editor1.join().unwrap();
    editor2.join().unwrap();

    // 关闭发送端——所有 Sender 被 drop 后，Receiver 的 for 循环会自动结束
    // tx 在这里被 drop（因为 tx 是主线程最后一个 sender）

    println!("所有编辑者已完成工作");
    receiver.join().unwrap();
    println!("=== 编辑会话结束 ===");
}
// 注意：tx 离开作用域时自动 drop，通道关闭
// 如果需要显式关闭，可以用 drop(tx)

=== 协作文档编辑器 ===
[服务器] 系统 插入了: 自动保存文档
[服务器] Alice 插入了: 第一章：Rust 简介
[服务器] Bob 插入了: Rust 是一门系统编程语言
[服务器] Alice 应用了格式: 标题加粗
[服务器] Bob 删除了第1行
所有编辑者已完成工作
[服务器] 通道关闭，停止接收
=== 编辑会话结束 ===

// ============================================
// 演示：Arc<Mutex<T>> 在多个线程间安全共享数据
// 模拟：多个工人同时修改一个共享计数器
// ============================================

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

fn main() {
    println!("=== 共享计数器演示 ===");

    // 用 Arc<Mutex<i32>> 包裹共享数据
    let counter = Arc::new(Mutex::new(0i32));
    let mut handles = vec![];

    // --- 启动 5 个线程，每个线程对计数器加 10 ---
    for id in 0..5 {
        let counter_clone = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            for _ in 0..10 {
                // lock() 获取互斥锁——如果锁被其他线程持有，当前线程会阻塞等待
                let mut num = counter_clone.lock().unwrap();
                *num += 1;
                println!("[工人{}] 当前计数: {}", id, *num);
                // 锁在离开作用域时自动释放
            }
            println!("[工人{}] 工作完成", id);
        });
        handles.push(handle);
    }

    // --- 等待所有线程结束 ---
    for handle in handles {
        handle.join().unwrap();
    }

    // --- 读取最终结果 ---
    let final_count = counter.lock().unwrap();
    println!("\n=== 最终结果 ===");
    println!("计数器最终值: {}", *final_count);
    println!("预期值: {} (5个线程 × 10次)", 5 * 10);
}

=== 共享计数器演示 ===
[工人0] 当前计数: 1
[工人0] 当前计数: 2
[工人1] 当前计数: 3
[工人1] 当前计数: 4
[工人0] 当前计数: 5
...（中间输出因线程调度而不同）
[工人4] 当前计数: 50
[工人4] 工作完成

=== 最终结果 ===
计数器最终值: 50
预期值: 50 (5个线程 × 10次)

// ============================================
// 生产者-消费者：多生产者 + 单消费者
// ============================================

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

fn main() {
    let (tx, rx) = mpsc::channel();
    let tx2 = tx.clone();

    thread::spawn(move || {
        let items = vec!["苹果", "香蕉", "橙子"];
        for item in items {
            tx.send(format!("生产者1: {}", item)).unwrap();
            thread::sleep(Duration::from_millis(100));
        }
        println!("生产者1 完成");
    });

    thread::spawn(move || {
        let items = vec!["西瓜", "葡萄"];
        for item in items {
            tx2.send(format!("生产者2: {}", item)).unwrap();
            thread::sleep(Duration::from_millis(150));
        }
        println!("生产者2 完成");
    });

    println!("=== 消费者接收 ===");
    for msg in rx {
        println!("  收到: {}", msg);
    }
    println!("所有生产者已退出，消费者结束");
}

=== 消费者接收 ===
  收到: 生产者1: 苹果
  收到: 生产者2: 西瓜
  收到: 生产者1: 香蕉
  收到: 生产者1: 橙子
  收到: 生产者2: 葡萄
生产者1 完成
生产者2 完成
所有生产者已退出，消费者结束

// ============================================
// 并行计算：多线程分片求和
// ============================================

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

fn parallel_sum(data: &[i64], num_threads: usize) -> i64 {
    let chunk_size = (data.len() + num_threads - 1) / num_threads;
    let result = Arc::new(Mutex::new(0i64));
    let mut handles = Vec::new();

    for i in 0..num_threads {
        let chunk = data[i * chunk_size..(i * chunk_size + chunk_size).min(data.len())].to_vec();
        let result = Arc::clone(&result);
        handles.push(thread::spawn(move || {
            let partial: i64 = chunk.iter().sum();
            *result.lock().unwrap() += partial;
            partial
        }));
    }

    let mut partials = Vec::new();
    for handle in handles {
        partials.push(handle.join().unwrap());
    }

    println!("各线程部分和: {:?}", partials);
    *result.lock().unwrap()
}

fn main() {
    let data: Vec<i64> = (1..=1000).collect();
    let sequential_sum: i64 = data.iter().sum();

    println!("=== 串行求和 ===");
    println!("1 到 1000 的和: {}", sequential_sum);

    println!("\n=== 并行求和 (4 线程) ===");
    let parallel_result = parallel_sum(&data, 4);
    println!("并行求和结果: {}", parallel_result);
    assert_eq!(sequential_sum, parallel_result);

    let data2: Vec<i64> = (1..=10_000_000).collect();
    let start = std::time::Instant::now();
    let _ = data2.iter().sum::<i64>();
    let seq_time = start.elapsed();

    let start = std::time::Instant::now();
    let _ = parallel_sum(&data2, 8);
    let par_time = start.elapsed();

    println!("\n=== 10M 数据性能对比 ===");
    println!("串行: {:?}", seq_time);
    println!("并行 (8线程): {:?}", par_time);
}

=== 串行求和 ===
1 到 1000 的和: 500500

=== 并行求和 (4 线程) ===
各线程部分和: [78126, 218874, 109374, 94126]
并行求和结果: 500500

=== 10M 数据性能对比 ===
串行: [时间]
并行 (8线程): [时间]