05-并发模型深度解析

并发模型深度解析

本章目标: 深入理解 Rust 并发模型,掌握 Send/Sync、Arc/Mutex、Channel 和线程安全编程

学习时间: 5-7 天

前置知识: 完成 04-类型系统与Trait.md

---

Rust 并发的核心承诺

“无畏并发”(Fearless Concurrency) —— Rust 在编译期保证并发代码不会出现数据竞争。

Go 的并发:   运行时检测数据竞争 (race detector)
Rust 的并发: 编译期拒绝有数据竞争的代码

---

1. Send 和 Sync: 线程安全基石

Send Trait

// Send 表示类型可以安全地跨线程转移所有权
pub unsafe trait Send {}
 
// 自动实现 Send 的类型:
// i32, f64, bool, String, Vec<T> (T: Send), ...
 
// 不实现 Send 的类型:
// Rc<T> - 非原子引用计数
// *mut T - 裸指针
// MutexGuard (不能在另一个线程 drop)

use std::thread;
use std::rc::Rc;
use std::sync::Arc;
 
fn send_demo() {
    let s = String::from("hello");  // String: Send
    
    let handle = thread::spawn(move || {
        println!("{}", s);  // s 移动到新线程, OK
    });
    
    handle.join().unwrap();
    
    // Rc 不是 Send
    let rc = Rc::new(5);
    // thread::spawn(move || println!("{}", rc));  // 编译错误!
    
    // Arc 是 Send
    let arc = Arc::new(5);
    let arc2 = Arc::clone(&arc);
    thread::spawn(move || println!("{}", arc2)).join().unwrap();
}

Sync Trait

// Sync 表示类型可以被多个线程同时不可变访问
// T: Sync 等价于 &T: Send
pub unsafe trait Sync {}
 
// 不实现 Sync 的类型:
// Cell<T> - 没有同步保证的内部可变性
// RefCell<T> - 运行时借用检查, 非线程安全
// Rc<T>

use std::sync::{Arc, Mutex};
 
// Arc<T> 是 Send + Sync (如果 T: Send + Sync)
fn sync_demo() {
    let data = Arc::new(vec![1, 2, 3]);
    
    let mut handles = vec![];
    for i in 0..3 {
        let data = Arc::clone(&data);
        handles.push(thread::spawn(move || {
            // 多个线程同时读 (Sync)
            println!("Thread {}: data[{}] = {}", i, i, data[i]);
        }));
    }
    
    for h in handles {
        h.join().unwrap();
    }
}

Send + Sync 关系图

          T: Send?          T: Sync?
                |                |
         可跨线程移动        可跨线程共享读
                |                |
    Arc<T>: if T: Send+Sync → Send+Sync
    Mutex<T>: if T: Send → Send+Sync
    Rc<T>: Never Send, Never Sync
    Cell<T>: Send (if T: Send), Never Sync

---

2. 线程和共享状态

基础线程

use std::thread;
use std::time::Duration;
 
fn basic_threads() {
    let handle = thread::spawn(|| {
        for i in 0..5 {
            println!("spawned thread: {}", i);
            thread::sleep(Duration::from_millis(1));
        }
    });
    
    for i in 0..5 {
        println!("main thread: {}", i);
        thread::sleep(Duration::from_millis(1));
    }
    
    handle.join().unwrap();  // 等待线程完成
}

Arc<Mutex>: 共享可变状态

use std::sync::{Arc, Mutex};
use std::thread;
 
fn shared_counter() {
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];
    
    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();
            *num += 1;
        });  // MutexGuard drop, 锁自动释放
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("Result: {}", *counter.lock().unwrap());  // 10
}

Mutex 内部工作原理

Arc<Mutex<T>> 内存布局:
┌──────────────────────────────────┐
│ Arc 控制块                        │
│   ref_count: AtomicUsize         │
├──────────────────────────────────┤
│ Mutex<T>                         │
│   locked: AtomicBool (或系统锁)  │
│   data: UnsafeCell<T>            │
└──────────────────────────────────┘

lock() 返回 MutexGuard:
- 实现 Deref/DerefMut → 访问内部数据
- 实现 Drop → 释放锁

死锁预防

use std::sync::{Arc, Mutex};
use std::thread;
 
// 死锁示例 (不要这样写!)
fn deadlock_example() {
    let lock1 = Arc::new(Mutex::new(1));
    let lock2 = Arc::new(Mutex::new(2));
    
    let l1 = Arc::clone(&lock1);
    let l2 = Arc::clone(&lock2);
    let t1 = thread::spawn(move || {
        let _a = l1.lock().unwrap();     // 获取 lock1
        thread::sleep(std::time::Duration::from_millis(1));
        let _b = l2.lock().unwrap();     // 等待 lock2 ← 可能死锁
    });
    
    let t2 = thread::spawn(move || {
        let _b = lock2.lock().unwrap();  // 获取 lock2
        thread::sleep(std::time::Duration::from_millis(1));
        let _a = lock1.lock().unwrap();  // 等待 lock1 ← 可能死锁
    });
    
    // 解决: 始终按相同顺序获取锁
}
 
// 正确做法: 锁排序
fn lock_ordering() {
    let lock1 = Arc::new(Mutex::new(1));
    let lock2 = Arc::new(Mutex::new(2));
    
    // 所有线程都先获取 lock1, 再获取 lock2
    fn get_both(l1: &Mutex<i32>, l2: &Mutex<i32>) -> (i32, i32) {
        let a = l1.lock().unwrap();
        let b = l2.lock().unwrap();
        (*a, *b)
    }
}

---

3. Channel: 消息传递

mpsc Channel (多生产者, 单消费者)

use std::sync::mpsc;
use std::thread;
 
fn basic_channel() {
    let (tx, rx) = mpsc::channel();
    
    thread::spawn(move || {
        let val = String::from("hello");
        tx.send(val).unwrap();
        // val 已被发送(移动), 不能再使用
        // println!("{}", val);  // 编译错误!
    });
    
    let received = rx.recv().unwrap();  // 阻塞等待消息
    println!("Got: {}", received);
}
 
// 多个生产者
fn multiple_producers() {
    let (tx, rx) = mpsc::channel();
    
    for i in 0..5 {
        let tx = tx.clone();  // 克隆发送端
        thread::spawn(move || {
            tx.send(i * i).unwrap();
        });
    }
    
    drop(tx);  // 关闭原始发送端
    
    // rx 会在所有 tx 都 drop 后关闭
    for received in rx {  // 迭代接收
        println!("Got: {}", received);
    }
}

Go Channel vs Rust Channel 对比

Go Channel:                 Rust mpsc Channel:
- 内置语法 (<-)             - 标准库 API
- 可以是有缓冲或无缓冲      - 有缓冲(sync_channel)和无缓冲
- 多生产者多消费者          - 多生产者单消费者(mpsc)
- select 语句              - 使用 crossbeam 的 select!
- GC 管理                  - 所有权管理

有缓冲 Channel

use std::sync::mpsc;
 
fn buffered_channel() {
    let (tx, rx) = mpsc::sync_channel(5);  // 缓冲区大小 5
    
    // 发送不会阻塞 (缓冲区未满)
    for i in 0..5 {
        tx.send(i).unwrap();
    }
    
    // 第 6 个会阻塞直到有空间
    // tx.send(5).unwrap();  // 会阻塞
    
    for v in rx.iter() {
        println!("{}", v);
    }
}

---

4. 并行计算: Rayon

// Rayon: 数据并行库
// cargo add rayon
 
use rayon::prelude::*;
 
fn parallel_computation() {
    let numbers: Vec<i64> = (1..=1_000_000).collect();
    
    // 顺序计算
    let sum_seq: i64 = numbers.iter().sum();
    
    // 并行计算 (几乎零代码改动!)
    let sum_par: i64 = numbers.par_iter().sum();
    
    assert_eq!(sum_seq, sum_par);
    
    // 并行 map + filter
    let result: Vec<i64> = numbers.par_iter()
        .filter(|&&x| x % 2 == 0)
        .map(|&x| x * x)
        .collect();
    
    println!("偶数平方的数量: {}", result.len());
}
 
// 并行排序
fn parallel_sort() {
    let mut data: Vec<i32> = (0..1_000_000).rev().collect();
    data.par_sort();  // 并行排序
    data.par_sort_by(|a, b| b.cmp(a));  // 并行逆序排序
}

---

5. 原子操作

use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering};
use std::sync::Arc;
use std::thread;
 
fn atomic_counter() {
    let counter = Arc::new(AtomicUsize::new(0));
    let mut handles = vec![];
    
    for _ in 0..1000 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            // 原子操作, 无需 Mutex
            counter.fetch_add(1, Ordering::Relaxed);
        });
        handles.push(handle);
    }
    
    for h in handles {
        h.join().unwrap();
    }
    
    println!("Counter: {}", counter.load(Ordering::Relaxed));  // 1000
}
 
// 原子标志
fn atomic_flag_demo() {
    let running = Arc::new(AtomicBool::new(true));
    let running_clone = Arc::clone(&running);
    
    let handle = thread::spawn(move || {
        let mut count = 0;
        while running_clone.load(Ordering::Acquire) {
            count += 1;
            if count % 1000000 == 0 {
                println!("Still running...");
            }
        }
        println!("Thread stopped after {} iterations", count);
    });
    
    thread::sleep(std::time::Duration::from_millis(10));
    running.store(false, Ordering::Release);  // 发送停止信号
    
    handle.join().unwrap();
}

内存序 (Memory Ordering)

// Ordering 从弱到强:
// Relaxed   - 只保证操作本身是原子的, 无顺序保证
// Acquire   - 读操作: 本次读之后的操作不会重排到读之前
// Release   - 写操作: 本次写之前的操作不会重排到写之后
// AcqRel    - 同时有 Acquire 和 Release 语义
// SeqCst    - 全序一致性, 最强但最慢
 
// 典型用法:
// 引用计数: Acquire/Release (Arc 内部使用)
// 标志位: AcqRel 或 SeqCst
// 简单计数器: Relaxed (不需要顺序保证)

---

6. 读写锁 (RwLock)

use std::sync::{Arc, RwLock};
use std::thread;
 
fn rwlock_demo() {
    let data = Arc::new(RwLock::new(vec![1, 2, 3]));
    let mut handles = vec![];
    
    // 多个读者
    for i in 0..5 {
        let data = Arc::clone(&data);
        handles.push(thread::spawn(move || {
            let read = data.read().unwrap();  // 获取读锁
            println!("Reader {}: {:?}", i, *read);
        }));
    }
    
    // 单个写者
    {
        let data = Arc::clone(&data);
        handles.push(thread::spawn(move || {
            let mut write = data.write().unwrap();  // 获取写锁
            write.push(4);
            println!("Writer appended 4");
        }));
    }
    
    for h in handles {
        h.join().unwrap();
    }
    
    println!("Final: {:?}", *data.read().unwrap());
}

---

7. 并发模式

工作池模式 (Worker Pool)

use std::sync::{Arc, Mutex};
use std::sync::mpsc;
use std::thread;
 
type Job = Box<dyn FnOnce() + Send + 'static>;
 
pub struct ThreadPool {
    workers: Vec<thread::JoinHandle<()>>,
    sender: mpsc::Sender<Job>,
}
 
impl ThreadPool {
    pub fn new(size: usize) -> Self {
        let (sender, receiver) = mpsc::channel::<Job>();
        let receiver = Arc::new(Mutex::new(receiver));
        
        let workers = (0..size).map(|id| {
            let receiver = Arc::clone(&receiver);
            thread::spawn(move || {
                loop {
                    let job = receiver.lock().unwrap().recv();
                    match job {
                        Ok(job) => {
                            println!("Worker {} executing job", id);
                            job();
                        }
                        Err(_) => {
                            println!("Worker {} stopping", id);
                            break;
                        }
                    }
                }
            })
        }).collect();
        
        ThreadPool { workers, sender }
    }
    
    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        self.sender.send(Box::new(f)).unwrap();
    }
}
 
impl Drop for ThreadPool {
    fn drop(&mut self) {
        drop(self.sender.clone());  // 关闭 channel
        // workers 会在 channel 关闭后自然退出
    }
}
 
fn main() {
    let pool = ThreadPool::new(4);
    
    for i in 0..8 {
        pool.execute(move || {
            println!("Task {} running on thread {:?}", i, thread::current().id());
            thread::sleep(std::time::Duration::from_millis(100));
        });
    }
    
    thread::sleep(std::time::Duration::from_millis(1000));
}

Barrier: 屏障同步

use std::sync::{Arc, Barrier};
use std::thread;
 
fn barrier_demo() {
    let n = 5;
    let barrier = Arc::new(Barrier::new(n));
    let mut handles = vec![];
    
    for i in 0..n {
        let b = Arc::clone(&barrier);
        handles.push(thread::spawn(move || {
            println!("Thread {} starting phase 1", i);
            
            // 模拟不同速度的第一阶段
            thread::sleep(std::time::Duration::from_millis(i as u64 * 10));
            
            println!("Thread {} waiting at barrier", i);
            b.wait();  // 所有线程都到达才继续
            
            println!("Thread {} starting phase 2", i);
        }));
    }
    
    for h in handles {
        h.join().unwrap();
    }
}

---

8. 实战: 并行文件处理

use std::fs;
use std::io::{self, BufRead};
use std::path::Path;
use std::sync::{Arc, Mutex};
use std::thread;
 
fn count_words_in_files(files: Vec<String>) -> std::collections::HashMap<String, usize> {
    let results = Arc::new(Mutex::new(std::collections::HashMap::new()));
    let mut handles = vec![];
    
    for file in files {
        let results = Arc::clone(&results);
        let handle = thread::spawn(move || {
            if let Ok(f) = fs::File::open(&file) {
                let reader = io::BufReader::new(f);
                let mut local_counts: std::collections::HashMap<String, usize> = 
                    std::collections::HashMap::new();
                
                for line in reader.lines().flatten() {
                    for word in line.split_whitespace() {
                        *local_counts.entry(word.to_lowercase()).or_insert(0) += 1;
                    }
                }
                
                // 合并到总结果
                let mut global = results.lock().unwrap();
                for (word, count) in local_counts {
                    *global.entry(word).or_insert(0) += count;
                }
            }
        });
        handles.push(handle);
    }
    
    for h in handles {
        h.join().unwrap();
    }
    
    Arc::try_unwrap(results)
        .expect("Still has references")
        .into_inner()
        .unwrap()
}

---

练习题

练习 1 (基础): Send/Sync 判断

以下哪些类型是 Send? 哪些是 Sync?

// 判断:
// a) Rc<i32>
// b) Arc<i32>
// c) Mutex<i32>
// d) Cell<i32>
// e) RefCell<i32>
// f) Arc<Mutex<i32>>
// g) *const i32

答案:

a) Rc<i32>:          不是 Send, 不是 Sync (引用计数非原子)
b) Arc<i32>:         是 Send, 是 Sync (原子引用计数)
c) Mutex<i32>:       是 Send, 是 Sync (提供同步保证)
d) Cell<i32>:        是 Send (如果 T: Send), 不是 Sync
e) RefCell<i32>:     是 Send (如果 T: Send), 不是 Sync
f) Arc<Mutex<i32>>:  是 Send, 是 Sync
g) *const i32:       不是 Send, 不是 Sync (裸指针)

---

练习 2 (基础): 并发计数器

实现一个线程安全的计数器:

use std::sync::{Arc, Mutex};
use std::thread;
 
fn main() {
    let counter = Arc::new(Mutex::new(HashMap::<String, usize>::new()));
    let mut handles = vec![];
    
    let words = vec!["apple", "banana", "apple", "cherry", "banana", "apple"];
    
    for word in words {
        let counter = Arc::clone(&counter);
        handles.push(thread::spawn(move || {
            let mut map = counter.lock().unwrap();
            *map.entry(word.to_string()).or_insert(0) += 1;
        }));
    }
    
    for h in handles {
        h.join().unwrap();
    }
    
    let result = counter.lock().unwrap();
    println!("{:?}", *result);
}

---

练习 3 (中等): Pipeline 模式

实现生产者-消费者 pipeline:

use std::sync::mpsc;
use std::thread;
 
fn pipeline() {
    // 阶段 1: 生成数据
    let (tx1, rx1) = mpsc::channel();
    thread::spawn(move || {
        for i in 0..10 {
            tx1.send(i).unwrap();
        }
    });
    
    // 阶段 2: 过滤偶数
    let (tx2, rx2) = mpsc::channel();
    thread::spawn(move || {
        for item in rx1 {
            if item % 2 == 0 {
                tx2.send(item).unwrap();
            }
        }
    });
    
    // 阶段 3: 平方
    let (tx3, rx3) = mpsc::channel();
    thread::spawn(move || {
        for item in rx2 {
            tx3.send(item * item).unwrap();
        }
    });
    
    // 收集结果
    let results: Vec<i32> = rx3.iter().collect();
    println!("{:?}", results);  // [0, 4, 16, 36, 64]
}

---

练习 4 (中等): 并行矩阵乘法

use std::sync::{Arc, Mutex};
use std::thread;
 
type Matrix = Vec<Vec<f64>>;
 
fn parallel_matmul(a: &Matrix, b: &Matrix, n_threads: usize) -> Matrix {
    let rows = a.len();
    let cols = b[0].len();
    let inner = b.len();
    
    let b = Arc::new(b.clone());
    let result = Arc::new(Mutex::new(vec![vec![0.0f64; cols]; rows]));
    
    let rows_per_thread = (rows + n_threads - 1) / n_threads;
    let mut handles = vec![];
    
    for t in 0..n_threads {
        let b = Arc::clone(&b);
        let result = Arc::clone(&result);
        let a_rows: Vec<Vec<f64>> = a[t * rows_per_thread..rows.min((t + 1) * rows_per_thread)]
            .to_vec();
        let row_offset = t * rows_per_thread;
        
        handles.push(thread::spawn(move || {
            let mut local: Vec<(usize, usize, f64)> = vec![];
            
            for (local_i, row) in a_rows.iter().enumerate() {
                let i = row_offset + local_i;
                for j in 0..cols {
                    let sum: f64 = (0..inner).map(|k| row[k] * b[k][j]).sum();
                    local.push((i, j, sum));
                }
            }
            
            let mut r = result.lock().unwrap();
            for (i, j, v) in local {
                r[i][j] = v;
            }
        }));
    }
    
    for h in handles {
        h.join().unwrap();
    }
    
    Arc::try_unwrap(result).unwrap().into_inner().unwrap()
}
 
fn main() {
    let a = vec![vec![1.0, 2.0], vec![3.0, 4.0]];
    let b = vec![vec![5.0, 6.0], vec![7.0, 8.0]];
    
    let c = parallel_matmul(&a, &b, 2);
    
    for row in &c {
        println!("{:?}", row);
    }
    // [19.0, 22.0]
    // [43.0, 50.0]
}

---

练习 5 (中等): 并发限速器

实现一个限制并发数的执行器:

use std::sync::{Arc, Mutex, Condvar};
use std::thread;
 
struct Semaphore {
    count: Mutex<usize>,
    cond: Condvar,
}
 
impl Semaphore {
    fn new(count: usize) -> Self {
        Semaphore {
            count: Mutex::new(count),
            cond: Condvar::new(),
        }
    }
    
    fn acquire(&self) {
        let mut count = self.count.lock().unwrap();
        while *count == 0 {
            count = self.cond.wait(count).unwrap();
        }
        *count -= 1;
    }
    
    fn release(&self) {
        *self.count.lock().unwrap() += 1;
        self.cond.notify_one();
    }
}
 
fn main() {
    let sem = Arc::new(Semaphore::new(3));  // 最多 3 个并发
    let mut handles = vec![];
    
    for i in 0..10 {
        let sem = Arc::clone(&sem);
        handles.push(thread::spawn(move || {
            sem.acquire();
            println!("Task {} running", i);
            thread::sleep(std::time::Duration::from_millis(100));
            println!("Task {} done", i);
            sem.release();
        }));
    }
    
    for h in handles {
        h.join().unwrap();
    }
}

---

练习 6 (挑战): 无锁队列 (使用原子操作)

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;
 
// 简单的单生产者单消费者无锁队列
struct SPSCQueue<T> {
    buffer: Vec<Option<T>>,
    head: AtomicUsize,
    tail: AtomicUsize,
    capacity: usize,
}
 
unsafe impl<T: Send> Send for SPSCQueue<T> {}
 
impl<T> SPSCQueue<T> {
    fn new(capacity: usize) -> Self {
        let mut buffer = Vec::with_capacity(capacity);
        for _ in 0..capacity {
            buffer.push(None);
        }
        SPSCQueue {
            buffer,
            head: AtomicUsize::new(0),
            tail: AtomicUsize::new(0),
            capacity,
        }
    }
    
    fn push(&self, item: T) -> bool {
        let tail = self.tail.load(Ordering::Relaxed);
        let next_tail = (tail + 1) % self.capacity;
        
        if next_tail == self.head.load(Ordering::Acquire) {
            return false;  // 队列满
        }
        
        unsafe { 
            let ptr = self.buffer.as_ptr().add(tail) as *mut Option<T>;
            ptr.write(Some(item));
        }
        
        self.tail.store(next_tail, Ordering::Release);
        true
    }
    
    fn pop(&self) -> Option<T> {
        let head = self.head.load(Ordering::Relaxed);
        
        if head == self.tail.load(Ordering::Acquire) {
            return None;  // 队列空
        }
        
        let item = unsafe {
            let ptr = self.buffer.as_ptr().add(head) as *mut Option<T>;
            ptr.replace(None)
        };
        
        self.head.store((head + 1) % self.capacity, Ordering::Release);
        item
    }
}

---

练习 7 (挑战): 并发下载器

实现一个简单的并发下载器(模拟):

use std::sync::{Arc, Mutex};
use std::thread;
use std::time::Duration;
use std::collections::HashMap;
 
#[derive(Debug, Clone)]
enum DownloadStatus {
    Pending,
    Downloading(u8),  // 百分比
    Completed,
    Failed(String),
}
 
struct Downloader {
    tasks: Arc<Mutex<HashMap<String, DownloadStatus>>>,
    max_concurrent: usize,
}
 
impl Downloader {
    fn new(max_concurrent: usize) -> Self {
        Downloader {
            tasks: Arc::new(Mutex::new(HashMap::new())),
            max_concurrent,
        }
    }
    
    fn download(&self, urls: Vec<String>) {
        use std::sync::mpsc;
        let (tx, rx) = mpsc::channel::<String>();
        let semaphore = Arc::new((Mutex::new(0usize), std::sync::Condvar::new()));
        
        for url in &urls {
            self.tasks.lock().unwrap().insert(url.clone(), DownloadStatus::Pending);
        }
        
        let mut handles = vec![];
        
        for url in urls {
            let tasks = Arc::clone(&self.tasks);
            let tx = tx.clone();
            let sem = Arc::clone(&semaphore);
            let max = self.max_concurrent;
            
            let handle = thread::spawn(move || {
                // 限流
                {
                    let (lock, cvar) = &*sem;
                    let mut count = lock.lock().unwrap();
                    while *count >= max {
                        count = cvar.wait(count).unwrap();
                    }
                    *count += 1;
                }
                
                // 模拟下载
                for progress in (0..=100).step_by(20) {
                    tasks.lock().unwrap().insert(
                        url.clone(), 
                        DownloadStatus::Downloading(progress)
                    );
                    thread::sleep(Duration::from_millis(50));
                }
                
                tasks.lock().unwrap().insert(url.clone(), DownloadStatus::Completed);
                
                // 释放信号量
                {
                    let (lock, cvar) = &*sem;
                    *lock.lock().unwrap() -= 1;
                    cvar.notify_one();
                }
                
                tx.send(url).unwrap();
            });
            handles.push(handle);
        }
        
        drop(tx);
        
        for completed_url in rx {
            println!("Completed: {}", completed_url);
        }
        
        for h in handles {
            h.join().unwrap();
        }
    }
    
    fn status(&self) -> HashMap<String, DownloadStatus> {
        self.tasks.lock().unwrap().clone()
    }
}
 
fn main() {
    let downloader = Downloader::new(3); // 最多3个并发
    
    let urls = vec![
        "http://example.com/file1.zip".to_string(),
        "http://example.com/file2.zip".to_string(),
        "http://example.com/file3.zip".to_string(),
        "http://example.com/file4.zip".to_string(),
        "http://example.com/file5.zip".to_string(),
    ];
    
    downloader.download(urls);
    
    let status = downloader.status();
    println!("\nFinal status:");
    for (url, status) in &status {
        println!("{}: {:?}", url, status);
    }
}

---

练习 8 (综合): 生产者-消费者与背压

use std::sync::{Arc, Mutex, Condvar};
use std::sync::mpsc;
use std::thread;
 
// 带背压的有界 channel
struct BoundedChannel<T> {
    inner: Arc<(Mutex<BoundedChannelInner<T>>, Condvar, Condvar)>,
}
 
struct BoundedChannelInner<T> {
    buffer: std::collections::VecDeque<T>,
    capacity: usize,
    closed: bool,
}
 
impl<T: Send + 'static> BoundedChannel<T> {
    fn new(capacity: usize) -> (Self, Self) {
        let inner = Arc::new((
            Mutex::new(BoundedChannelInner {
                buffer: std::collections::VecDeque::new(),
                capacity,
                closed: false,
            }),
            Condvar::new(),  // not_full
            Condvar::new(),  // not_empty
        ));
        (BoundedChannel { inner: inner.clone() }, BoundedChannel { inner })
    }
    
    fn send(&self, item: T) -> bool {
        let (lock, not_full, not_empty) = &*self.inner;
        let mut guard = lock.lock().unwrap();
        
        while guard.buffer.len() >= guard.capacity && !guard.closed {
            guard = not_full.wait(guard).unwrap();
        }
        
        if guard.closed { return false; }
        
        guard.buffer.push_back(item);
        not_empty.notify_one();
        true
    }
    
    fn recv(&self) -> Option<T> {
        let (lock, not_full, not_empty) = &*self.inner;
        let mut guard = lock.lock().unwrap();
        
        while guard.buffer.is_empty() && !guard.closed {
            guard = not_empty.wait(guard).unwrap();
        }
        
        let item = guard.buffer.pop_front();
        if item.is_some() {
            not_full.notify_one();
        }
        item
    }
}

---

练习 9 (综合): 并发 KV 存储

use std::collections::HashMap;
use std::sync::{Arc, RwLock};
 
pub struct ConcurrentKV<K, V> {
    // 分片设计减少锁竞争
    shards: Vec<Arc<RwLock<HashMap<K, V>>>>,
    shard_count: usize,
}
 
impl<K, V> ConcurrentKV<K, V>
where
    K: std::hash::Hash + Eq + Clone,
    V: Clone,
{
    pub fn new(shard_count: usize) -> Self {
        let shards = (0..shard_count)
            .map(|_| Arc::new(RwLock::new(HashMap::new())))
            .collect();
        ConcurrentKV { shards, shard_count }
    }
    
    fn shard_index(&self, key: &K) -> usize {
        use std::collections::hash_map::DefaultHasher;
        use std::hash::{Hash, Hasher};
        
        let mut hasher = DefaultHasher::new();
        key.hash(&mut hasher);
        hasher.finish() as usize % self.shard_count
    }
    
    pub fn insert(&self, key: K, value: V) {
        let idx = self.shard_index(&key);
        self.shards[idx].write().unwrap().insert(key, value);
    }
    
    pub fn get(&self, key: &K) -> Option<V> {
        let idx = self.shard_index(key);
        self.shards[idx].read().unwrap().get(key).cloned()
    }
    
    pub fn remove(&self, key: &K) -> Option<V> {
        let idx = self.shard_index(key);
        self.shards[idx].write().unwrap().remove(key)
    }
}
 
fn main() {
    let kv = Arc::new(ConcurrentKV::new(16));
    let mut handles = vec![];
    
    for i in 0..100 {
        let kv = Arc::clone(&kv);
        handles.push(std::thread::spawn(move || {
            kv.insert(format!("key{}", i), i * i);
        }));
    }
    
    for h in handles {
        h.join().unwrap();
    }
    
    println!("key42: {:?}", kv.get(&"key42".to_string()));
}

---

练习 10 (综合): 对比 Go 和 Rust 的并发

分析以下场景,用 Rust 实现 Go 的 sync.WaitGroup:

use std::sync::{Arc, Mutex, Condvar};
 
pub struct WaitGroup {
    inner: Arc<(Mutex<i32>, Condvar)>,
}
 
impl WaitGroup {
    pub fn new() -> Self {
        WaitGroup {
            inner: Arc::new((Mutex::new(0), Condvar::new())),
        }
    }
    
    pub fn add(&self, delta: i32) {
        let (lock, cond) = &*self.inner;
        let mut count = lock.lock().unwrap();
        *count += delta;
        if *count == 0 {
            cond.notify_all();
        }
    }
    
    pub fn done(&self) {
        self.add(-1);
    }
    
    pub fn wait(&self) {
        let (lock, cond) = &*self.inner;
        let mut count = lock.lock().unwrap();
        while *count > 0 {
            count = cond.wait(count).unwrap();
        }
    }
    
    pub fn clone(&self) -> Self {
        WaitGroup { inner: Arc::clone(&self.inner) }
    }
}
 
fn main() {
    let wg = WaitGroup::new();
    
    for i in 0..5 {
        let wg = wg.clone();
        wg.add(1);
        std::thread::spawn(move || {
            println!("Task {} running", i);
            std::thread::sleep(std::time::Duration::from_millis(100));
            println!("Task {} done", i);
            wg.done();
        });
    }
    
    wg.wait();
    println!("All tasks completed");
}

---

小结

概念	要点
Send	可跨线程转移所有权
Sync	可跨线程共享引用
Arc	原子引用计数, 多线程共享
Mutex	互斥锁, 独占访问
RwLock	读写锁, 多读单写
mpsc	多生产者单消费者 channel
Rayon	数据并行, par_iter

下一步

• 完成本章所有练习题
• 阅读 06-Unsafe与链表实现.md
• 尝试用 rayon 优化一个计算密集任务

---

Rust 的并发安全由类型系统保证 —— 如果编译通过就说明没有数据竞争! 🦀