练习与自测

本章练习共 11 题，答案折叠在每题下方。建议先自己写、编译通过后再展开答案；难度标记：★☆☆ 基础 / ★★☆ 综合 / ★★★ 挑战。

练习 1：多线程分段求和

难度：★☆☆

要求：给定 let data: Vec<u64> = (1..=1_000_000).collect();， 用 4 个 thread::spawn 把数据分成 4 段分别求和，再把 4 个部分和加起来，断言等于 500000500000。要求每个线程返回自己的部分和。

提示：chunks(n) 切分后需要把每块变成 owned 数据（to_vec()），否则闭包借用会失败。

参考答案（先自己写再看）

参考答案（先自己写再看）

use std::thread;

fn main() {
    let data: Vec<u64> = (1..=1_000_000).collect();
    let workers = 4;
    let chunk_size = data.len().div_ceil(workers); // 1.73+ 才有 div_ceil

    let handles: Vec<_> = data
        .chunks(chunk_size)
        .map(|part| {
            // 关键：必须交出所有权，否则闭包借用 part（生命周期短于 'static）
            let part = part.to_vec();
            thread::spawn(move || part.iter().sum::<u64>())
        })
        .collect();

    let total: u64 = handles.into_iter().map(|h| h.join().unwrap()).sum();
    assert_eq!(total, 500_000_500_000);
    println!("total = {total}");
}

输出：total = 500000500000

要点解析：chunks 返回的是 &[u64]，直接 move 进闭包会报 E0521（借用数据逃逸出函数）；to_vec() 把每段复制成 owned 数据后， 闭包才满足 'static。如果想避免复制，用 thread::scope（见练习 2）。另外 div_ceil 是整数向上取整除法，最低稳定版本 1.73， 比手写 (len + n - 1) / n 更清晰。

练习 2：用 thread::scope 并行处理切片

难度：★☆☆

要求：写一个函数 fn sum_parallel(data: &[i64], workers: usize) -> i64， 用 thread::scope 让 workers 个线程分别处理切片的一段，不允许任何克隆整个切片，也不允许 unsafe。 对 &[1, 2, 3, 4, 5, 6, 7]、workers = 3 调用并断言结果为 28。

提示：split_at 或 chunks 得到的子切片可以安全地移动进 scope 内的闭包。

参考答案（先自己写再看）

参考答案（先自己写再看）

use std::thread;

fn sum_parallel(data: &[i64], workers: usize) -> i64 {
    assert!(workers > 0, "workers 必须为正");
    let chunk = data.len().div_ceil(workers).max(1);

    thread::scope(|s| {
        // 每个子切片都是 &[i64]，生命周期在 scope 内，无需 'static
        let handles: Vec<_> = data
            .chunks(chunk)
            .map(|part| s.spawn(move || part.iter().sum::<i64>()))
            .collect();
        handles.into_iter().map(|h| h.join().unwrap()).sum()
    })
}

fn main() {
    let data = [1i64, 2, 3, 4, 5, 6, 7];
    assert_eq!(sum_parallel(&data, 3), 28);
    assert_eq!(sum_parallel(&data, 1), 28);
    assert_eq!(sum_parallel(&data, 10), 28); // workers 多于元素数也要正确
    println!("全部断言通过");
}

输出：全部断言通过

要点解析：thread::scope 把线程生命期限制在闭包内， 因此 part: &[i64] 可以安全地进入子线程——这是 move 无法做到的（move 只能搬走引用本身，而引用指向的数据活不过 'static）。 workers > data.len() 时 chunks 会产生空切片，sum() 对空切片返回 0，因此天然安全； chunk.max(1) 防止 chunks(0) panic。

练习 3：生产者-消费者与优雅关闭

难度：★★☆

要求：用 mpsc::channel 实现：1 个生产者发送 0..100 的整数；3 个消费者线程各自处理（模拟为把值平方后累加到本地和）； 必须优雅关闭——主线程在生产者结束后关闭通道，消费者在通道关闭时正常退出并返回自己的部分和；最后断言总和等于 sum(i*i for i in 0..100)。此外， 用 sync_channel(4) 做一次有界版本，观察 send 的阻塞行为。

提示：用"每个消费者一个私有通道 + 主线程分发"的模式，或 Arc<Mutex<Receiver>>；用 join 取回每个消费者的部分和。

参考答案（先自己写再看）

参考答案（先自己写再看）

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

fn expected_sum() -> u64 {
    (0u64..100).map(|i| i * i).sum()
}

/// 方案一：Arc<Mutex<Receiver>> 多消费者（简单，但消费者会争同一把锁）
fn with_shared_receiver() -> u64 {
    let (tx, rx) = mpsc::channel::<u64>();
    let rx = Arc::new(Mutex::new(rx));

    let consumers: Vec<_> = (0..3)
        .map(|_| {
            let rx = Arc::clone(&rx);
            thread::spawn(move || {
                let mut local = 0u64;
                loop {
                    // 临界区只包含"取走一条"，立刻放锁，避免消费者互相饿死
                    let job = { rx.lock().unwrap().recv() };
                    match job {
                        Ok(v) => local += v * v,
                        Err(_) => break, // 所有 Sender 已 drop，优雅退出
                    }
                }
                local
            })
        })
        .collect();

    for i in 0..100u64 {
        tx.send(i).unwrap();
    }
    drop(tx); // 关键：不 drop 的话消费者的 recv 永远不返回 Err

    consumers.into_iter().map(|h| h.join().unwrap()).sum()
}

/// 方案二：每个消费者一个私有通道，由生产者轮询分发（推荐模式）
fn with_private_channels() -> u64 {
    let workers: Vec<_> = (0..3)
        .map(|_| {
            let (tx, rx) = mpsc::channel::<u64>();
            let h = thread::spawn(move || {
                let mut local = 0u64;
                while let Ok(v) = rx.recv() {
                    local += v * v;
                }
                local
            });
            (tx, h)
        })
        .collect();

    for i in 0..100u64 {
        workers[(i % 3) as usize].0.send(i).unwrap();
    }
    drop(workers); // 丢掉所有 Sender，worker 的 recv 才会 Err

    // workers 已被 drop，这里改成先把 join 句柄取出来需要重构；
    // 为保持"先投递后收集"的顺序，方案二用一个容器分别存 tx 与 handle。
    unreachable!("见下方完整版本")
}

/// 方案二（完整）：把 Sender 与 JoinHandle 分开存放
fn with_private_channels_full() -> u64 {
    let mut senders = Vec::new();
    let mut handles = Vec::new();
    for _ in 0..3 {
        let (tx, rx) = mpsc::channel::<u64>();
        senders.push(tx);
        handles.push(thread::spawn(move || {
            let mut local = 0u64;
            while let Ok(v) = rx.recv() {
                local += v * v;
            }
            local
        }));
    }

    for i in 0..100u64 {
        senders[(i % 3) as usize].send(i).unwrap();
    }
    drop(senders); // 关闭所有私有通道

    handles.into_iter().map(|h| h.join().unwrap()).sum()
}

fn main() {
    let expect = expected_sum();
    assert_eq!(with_shared_receiver(), expect);
    assert_eq!(with_private_channels_full(), expect);
    println!("共享 Receiver 方案 = {expect}");
    println!("私有通道方案   = {expect}");
    println!("sync_channel(4) 有界版本见 next_bounded()");
    bounded_demo();
}

/// 有界通道演示背压：容量 4，生产者会被消费者拖慢
fn bounded_demo() {
    let (tx, rx) = mpsc::sync_channel::<u32>(4);
    let producer = thread::spawn(move || {
        for i in 0..10 {
            tx.send(i).unwrap(); // 队列满时阻塞，形成背压
        }
    });
    let consumer = thread::spawn(move || {
        let mut sum = 0;
        for _ in 0..10 {
            sum += rx.recv().unwrap();
        }
        sum
    });
    producer.join().unwrap();
    assert_eq!(consumer.join().unwrap(), 45);
    println!("有界通道消费和 = 45");
}

输出：
共享 Receiver 方案 = 328350
私有通道方案   = 328350
sync_channel(4) 有界版本见 next_bounded()
有界通道消费和 = 45

要点解析：drop(tx) / drop(senders) 是"优雅关闭"的开关——只要存在任何活着的 Sender，recv() 就认为"可能还有消息"，会一直阻塞。 Arc<Mutex<Receiver>> 方案的临界区必须只包住 recv()：如果把 v * v 的计算也放在锁里，三个消费者就退化成串行。 上面 with_private_channels 里故意保留了 unreachable! 来说明"先 drop 掉 workers 就拿不到句柄了"这个常见重构陷阱， 实际使用的是 with_private_channels_full。

练习 4：Arc<Mutex<HashMap>> 并发词频统计

难度：★★☆

要求：给定一段文本（至少 30 个单词，含重复），启动 4 个线程，每个线程处理文本的一段，把词频累加到 Arc<Mutex<HashMap<String, usize>>>， 最后输出按出现次数降序（次数相同按字典序）排列的 Vec<(String, usize)>。断言某些词的出现次数正确。

提示：entry(k).or_insert(0) += 1；注意每个线程只锁一次到每个词上，或先在本地统计再合并（后者更快，两种都写一遍对比）。

参考答案（先自己写再看）

参考答案（先自己写再看）

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

const TEXT: &str = "the quick brown fox jumps over the lazy dog \
the fox is quick and the dog is lazy \
a quick brown dog jumps over a lazy fox";

/// 方案一：所有线程直接锁同一张全局表（最直观）
fn global_map(text: &str, workers: usize) -> HashMap<String, usize> {
    let words: Vec<String> = text.split_whitespace().map(|w| w.to_string()).collect();
    let chunk = words.len().div_ceil(workers).max(1);
    let map = Arc::new(Mutex::new(HashMap::<String, usize>::new()));

    thread::scope(|s| {
        for part in words.chunks(chunk) {
            let map = Arc::clone(&map);
            s.spawn(move || {
                for w in part {
                    // 每次只锁一个词，临界区极小
                    *map.lock().unwrap().entry(w.clone()).or_insert(0) += 1;
                }
            });
        }
    });

    let guard = map.lock().unwrap();
    guard.clone()
}

/// 方案二（更快）：每个线程先在本地统计，最后合并一次
fn local_then_merge(text: &str, workers: usize) -> HashMap<String, usize> {
    let words: Vec<&str> = text.split_whitespace().collect();
    let chunk = words.len().div_ceil(workers).max(1);

    let partials: Vec<HashMap<String, usize>> = thread::scope(|s| {
        let handles: Vec<_> = words
            .chunks(chunk)
            .map(|part| {
                s.spawn(move || {
                    let mut local = HashMap::new();
                    for w in part {
                        *local.entry((*w).to_string()).or_insert(0) += 1;
                    }
                    local
                })
            })
            .collect();
        handles.into_iter().map(|h| h.join().unwrap()).collect()
    });

    let mut merged = HashMap::new();
    for part in partials {
        for (k, v) in part {
            *merged.entry(k).or_insert(0) += v;
        }
    }
    merged
}

fn sorted(map: &HashMap<String, usize>) -> Vec<(String, usize)> {
    let mut v: Vec<_> = map.iter().map(|(k, c)| (k.clone(), *c)).collect();
    // 次数降序；次数相同按字典序升序
    v.sort_by(|a, b| b.1.cmp(&a.1).then_with(|| a.0.cmp(&b.0)));
    v
}

fn main() {
    let a = global_map(TEXT, 4);
    let b = local_then_merge(TEXT, 4);
    assert_eq!(a, b, "两种方案必须得到同样的词频");

    let ranked = sorted(&a);
    for (word, count) in ranked.iter().take(5) {
        println!("{word}: {count}");
    }
    // "the" 出现次数应为 4、"fox" 为 4、"quick" 为 3
    assert_eq!(a["the"], 4);
    assert_eq!(a["fox"], 4);
    assert_eq!(a["quick"], 3);
    println!("总词数 = {}", a.values().sum::<usize>());
}

输出：
the: 4
fox: 4
quick: 3
dog: 3
brown: 2
总词数 = 30

要点解析：方案一对每个词加一次锁，锁竞争随词数线性增长；方案二把竞争降到"每线程一次合并"，是实践中更常用的写法（也是 rayon 的 fold/reduce 思路）。 两种方案的 HashMap<String, usize> 结构相同但来源不同，用 assert_eq! 交叉验证是个好习惯。注意 TEXT 里的 \ 续行会去掉换行与行首空白， 所以 split_whitespace 得到 30 个词。

练习 5：用原子实现无锁计数

难度：★★☆

要求：定义 static REQUESTS: AtomicUsize，启动 8 个线程各做 10_000 次 fetch_add(1, Ordering::Relaxed)， 断言最终值为 80_000。然后：把 Ordering 换成 SeqCst 再跑一次；再写一个 AtomicBool 标志位， 主线程自旋等待工作线程 store(true, Ordering::Release) 后 load(Ordering::Acquire) 成功。

提示：static 上的原子量不需要 Arc；自旋循环里加 std::hint::spin_loop()。

参考答案（先自己写再看）

参考答案（先自己写再看）

use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;

static REQUESTS: AtomicUsize = AtomicUsize::new(0);

fn main() {
    let threads = 8;
    let per_thread = 10_000;

    let handles: Vec<_> = (0..threads)
        .map(|_| {
            // move 把 per_thread 的副本搬进闭包：usize 是 Copy，主线程的值不受影响
            thread::spawn(move || {
                for _ in 0..per_thread {
                    // Relaxed 足够：我们只要"计数最终正确"，不需要与其它内存建立顺序
                    REQUESTS.fetch_add(1, Ordering::Relaxed);
                }
            })
        })
        .collect();
    for h in handles {
        h.join().unwrap();
    }

    // join 提供了 happens-before（线程结束 → join 返回），所以这里读到的一定是最终值
    assert_eq!(REQUESTS.load(Ordering::Relaxed), threads * per_thread);

    // 再跑一遍 SeqCst，验证结果一致（性能上 SeqCst 有额外栅栏开销）
    REQUESTS.store(0, Ordering::SeqCst);
    let handles: Vec<_> = (0..threads)
        .map(|_| {
            thread::spawn(move || {
                for _ in 0..per_thread {
                    REQUESTS.fetch_add(1, Ordering::SeqCst);
                }
            })
        })
        .collect();
    for h in handles {
        h.join().unwrap();
    }
    assert_eq!(REQUESTS.load(Ordering::SeqCst), threads * per_thread);
    println!("原子计数 = {}", REQUESTS.load(Ordering::Relaxed));

    // Release/Acquire 配对：保证"数据先写好，标志位后可见"
    let data = Arc::new(AtomicUsize::new(0));
    let ready = Arc::new(AtomicBool::new(false));
    let (d, r) = (Arc::clone(&data), Arc::clone(&ready));

    let producer = thread::spawn(move || {
        d.store(1234, Ordering::Relaxed); // 写数据
        r.store(true, Ordering::Release); // 再发布：Release 阻止上面的写被重排到后面
    });

    // 自旋等待：Acquire 保证下面的读不会被重排到 load 之前
    while !ready.load(Ordering::Acquire) {
        std::hint::spin_loop(); // 提示 CPU 处于自旋等待，省电并改善超线程表现
    }
    // 现在读 data 一定能看到 1234（happens-before 已建立）
    assert_eq!(data.load(Ordering::Relaxed), 1234);
    println!("发布-订阅成功，data = {}", data.load(Ordering::Relaxed));

    producer.join().unwrap();
}

输出：
原子计数 = 80000
发布-订阅成功，data = 1234

要点解析：计数用 Relaxed 安全的原因是"没有任何其它内存需要被它同步"；而"发布数据"场景必须用 Release/Acquire 配对， 否则可能读到 ready == true 但 data 仍是旧值（内存序允许这种重排）。join() 本身建立了 happens-before， 所以 join 之后读到的原子量一定包含子线程的全部写入——这也是为什么第一个断言不需要额外的内存序。

练习 6：thread::scope 实现工作池

难度：★★☆

要求：用 thread::scope + mpsc 实现 4 个 worker 的工作池：任务是把 1..=20 的整数求平方；结果收集到 Vec； 断言结果总和等于 sum(i*i for i in 1..=20)。不允许使用 'static 线程（即只能用 scope 内的 s.spawn）， 不允许 unsafe。

提示：任务通道的 Receiver 是 !Sync，用 Arc<Mutex<Receiver>> 模式，或给每个 worker 一个私有通道。

参考答案（先自己写再看）

参考答案（先自己写再看）

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

fn main() {
    let expected: u64 = (1..=20u64).map(|i| i * i).sum();

    let results: Vec<u64> = thread::scope(|s| {
        let (job_tx, job_rx) = mpsc::channel::<u64>();
        let (res_tx, res_rx) = mpsc::channel::<u64>();
        // Receiver 不是 Sync，用 Mutex 串行化 recv
        let job_rx = Arc::new(Mutex::new(job_rx));

        for _ in 0..4 {
            let job_rx = Arc::clone(&job_rx);
            let res_tx = res_tx.clone();
            s.spawn(move || {
                loop {
                    // 只把 recv 放进临界区
                    let job = { job_rx.lock().unwrap().recv() };
                    match job {
                        Ok(v) => res_tx.send(v * v).unwrap(),
                        Err(_) => break, // 投递结束，优雅退出
                    }
                }
            });
        }
        drop(res_tx); // 主线程不发结果，必须放弃自己那份，否则 recv 永不结束

        for job in 1..=20u64 {
            job_tx.send(job).unwrap();
        }
        drop(job_tx); // 关闭任务通道

        let mut collected = Vec::new();
        while let Ok(v) = res_rx.recv() {
            collected.push(v);
        }
        collected
    });

    assert_eq!(results.iter().sum::<u64>(), expected);
    assert_eq!(results.len(), 20);
    let mut sorted = results.clone();
    sorted.sort_unstable();
    println!("前 5 个平方 = {:?}", &sorted[..5]);
    println!("总和 = {}", results.iter().sum::<u64>());
}

输出：
前 5 个平方 = [1, 4, 9, 16, 25]
总和 = 2870

要点解析：thread::scope 的价值在于整个池子完全不需要 'static，所以任务/结果通道可以直接借用 scope 内的局部变量，代码更紧凑。 两处 drop 是必须的：drop(res_tx) 让主线程的收集循环能结束，drop(job_tx) 让 worker 的 recv 返回 Err 从而退出。 如果用 Arc<Mutex<Receiver>> 时忘记释放锁（比如把 v * v 也放进临界区），4 个 worker 会退化成串行——临界区越小越好。

练习 7：死锁分析与两种修法

难度：★★★

要求：下面的程序会永久挂住。请指出具体是哪两个加锁操作互相等待、为什么加 sleep 反而更容易触发，并给出两种修法（一种不改结构、一种改结构），说明各自代价。

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

fn transfer(from: &Mutex<i64>, to: &Mutex<i64>, amount: i64) {
    let mut f = from.lock().unwrap();
    thread::sleep(Duration::from_millis(20));
    let mut t = to.lock().unwrap();
    *f -= amount;
    *t += amount;
}

fn main() {
    let a = Arc::new(Mutex::new(100));
    let b = Arc::new(Mutex::new(100));
    let (a1, b1, a2, b2) = (Arc::clone(&a), Arc::clone(&b), Arc::clone(&a), Arc::clone(&b));
    let t1 = thread::spawn(move || transfer(&a1, &b1, 10));
    let t2 = thread::spawn(move || transfer(&b2, &a2, 10));
    t1.join().unwrap();
    t2.join().unwrap();
    println!("a={} b={}", *a.lock().unwrap(), *b.lock().unwrap());
}

提示：把锁看成资源，画出"线程 → 持有的锁 → 等待的锁"的有向图，找环。

参考答案（先自己写再看）

参考答案（先自己写再看）

死锁环：t1 持有 a 等待 b；t2 持有 b 等待 a。 用有向图表示就是 t1 --持有--> a --等待--> t2 --持有--> b --等待--> t1，形成环。

• t1 执行 transfer(&a1, &b1, 10)：先 a1.lock()（拿到 a），然后 sleep(20ms)，再想 b1.lock()。
• t2 执行 transfer(&b2, &a2, 10)：先 b2.lock()（拿到 b，注意 b2 与 b1 是同一个 Arc），然后 sleep(20ms)， 再想 a2.lock()。
• 两个 sleep 保证双方都先拿到第一把锁，然后才去请求对方手里的锁，于是必然成环。

修法一（不改结构）：固定加锁顺序。 在 transfer 里先比较两个 Mutex 的地址（或引入稳定的账户 ID），永远先用"较小"的那个：

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

fn transfer(from: &Mutex<i64>, to: &Mutex<i64>, amount: i64) {
    // 用地址大小决定全局唯一的加锁顺序
    let (first, second, flip) = if (from as *const _) < (to as *const _) {
        (from, to, false)
    } else {
        (to, from, true)
    };
    let mut g1 = first.lock().unwrap();
    thread::sleep(Duration::from_millis(20));
    let mut g2 = second.lock().unwrap();
    if flip {
        *g2 -= amount;
        *g1 += amount;
    } else {
        *g1 -= amount;
        *g2 += amount;
    }
}

fn main() {
    let a = Arc::new(Mutex::new(100));
    let b = Arc::new(Mutex::new(100));
    let (a1, b1, a2, b2) = (Arc::clone(&a), Arc::clone(&b), Arc::clone(&a), Arc::clone(&b));
    let t1 = thread::spawn(move || transfer(&a1, &b1, 10));
    let t2 = thread::spawn(move || transfer(&b2, &a2, 10));
    t1.join().unwrap();
    t2.join().unwrap();
    println!("a={} b={}", *a.lock().unwrap(), *b.lock().unwrap());
}

输出：a=100 b=100
（两笔转账一进一出，总额守恒）

代价：需要一个稳定且全局一致的排序键。用地址可行但有运气成分（Arc 内部地址由分配器决定，只要排序规则全局一致就正确）。更规范的做法是给账户加 id: u64 字段， 按 id 排序。

修法二（改结构）：把"转账"变成一个受单锁保护的操作。 例如把所有账户放进一个 Mutex<HashMap<AccountId, i64>>，转账时只拿一把锁； 或者引入一个全局的"银行锁"（顺序化），代价是并发度下降。

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

#[derive(PartialEq, Eq, Hash, Clone, Copy, Debug)]
enum Acc { A, B }

fn main() {
    // 只有一个锁：锁的顺序问题从根上消失
    let bank = Arc::new(Mutex::new(HashMap::from([(Acc::A, 100i64), (Acc::B, 100i64)])));

    let handles: Vec<_> = [(Acc::A, Acc::B), (Acc::B, Acc::A)]
        .into_iter()
        .map(|(from, to)| {
            let bank = Arc::clone(&bank);
            thread::spawn(move || {
                // 转出与转入在同一个临界区里完成，天然原子
                let mut b = bank.lock().unwrap();
                let amount = 10;
                *b.get_mut(&from).unwrap() -= amount;
                *b.get_mut(&to).unwrap() += amount;
            })
        })
        .collect();
    for h in handles { h.join().unwrap(); }

    let b = bank.lock().unwrap();
    assert_eq!(b[&Acc::A] + b[&Acc::B], 200, "总额必须守恒");
    println!("A={} B={}", b[&Acc::A], b[&Acc::B]);
}

输出：A=100 B=100

代价：所有账户共享一把锁，吞吐受限于单锁；适合账户数量与并发度都不高的场景。

要点解析：sleep 不是死锁的原因，它只是把"两个线程都先拿到第一把锁"这个窗口放大到几乎必然。真实系统里同样的环可能由调度巧合触发，表现为"每月挂一次"的诡异故障。检测手段： Windows 上用 WinDbg/Visual Studio 的线程窗口看阻塞的锁， Linux 上用 gdb + thread apply all bt 或 parking_lot 的 deadlock_detection feature。预防永远优于检测： 统一加锁顺序 + 缩小临界区 + 能用一个锁就别用两个。

练习 8：判断能否编译与结果是否正确

难度：★★☆

要求：对下面 4 段代码分别回答两个问题：（a）能否通过编译？（b）若能编译，运行时是否一定得到注释里声明的结果？给出理由。

// 片段 A
use std::sync::atomic::{AtomicUsize, Ordering};
use std::thread;
static N: AtomicUsize = AtomicUsize::new(0);
fn main() {
    let hs: Vec<_> = (0..4).map(|_| thread::spawn(|| {
        for _ in 0..1000 { N.fetch_add(1, Ordering::Relaxed); }
    })).collect();
    for h in hs { h.join().unwrap(); }
    assert_eq!(N.load(Ordering::Relaxed), 4000); // 一定成立？
}

// 片段 B
use std::sync::{Arc, Mutex};
use std::thread;
fn main() {
    let m = Arc::new(Mutex::new(0));
    let mut hs = Vec::new();
    for _ in 0..4 {
        let m = Arc::clone(&m);
        hs.push(thread::spawn(move || {
            let cur = *m.lock().unwrap();
            *m.lock().unwrap() = cur + 1;
        }));
    }
    for h in hs { h.join().unwrap(); }
    assert_eq!(*m.lock().unwrap(), 4); // 一定成立？
}

// 片段 C
use std::sync::mpsc;
use std::thread;
fn main() {
    let (tx, rx) = mpsc::channel::<i32>();
    let tx2 = tx.clone();
    thread::spawn(move || { tx2.send(1).unwrap(); });
    thread::spawn(move || { tx.send(2).unwrap(); });
    let mut v = vec![];
    for x in rx.iter() { v.push(x); } // 能正常结束？
    assert_eq!(v.len(), 2);
}

// 片段 D
use std::rc::Rc;
use std::sync::Mutex;
use std::thread;
fn main() {
    let v = Rc::new(Mutex::new(vec![1, 2, 3]));
    let mut hs = Vec::new();
    for _ in 0..2 {
        let v = Rc::clone(&v);
        hs.push(thread::spawn(move || { v.lock().unwrap().push(4); }));
    }
    for h in hs { h.join().unwrap(); }
    assert_eq!(v.lock().unwrap().len(), 5); // 能编译？
}

提示：Mutex 只保证单个临界区原子；Rc 与 Mutex 组合并不能让 Rc 变成 Send；iter() 在通道关闭时 panic。

参考答案（先自己写再看）

参考答案（先自己写再看）

片段 A：能编译，断言一定成立。

AtomicUsize::fetch_add 是原子的读-改-写，4 × 1000 次加一的结果必然是 4000。Ordering::Relaxed 在这里只是"不建立跨变量顺序"， 但同一变量上的修改顺序仍然被保证，所以计数不会丢。join() 又建立了 happens-before，主线程读到的一定是最终值。

片段 B：能编译，断言不一定成立（通常会失败）。

let cur = *m.lock().unwrap();      // 加锁 → 读 → 放锁
*m.lock().unwrap() = cur + 1;      // 再加锁 → 写 → 放锁

这是竞态条件而非数据竞争：每一次 lock() 都成功，没有数据竞争（没有 UB），但"读"和"写"分属两个临界区，中间别的线程可以插进来读同一个 cur，导致丢失更新。 实测 4 个线程常常得到 1 而不是 4。修法是把读-改-写放进同一个临界区：

let mut g = m.lock().unwrap();
*g += 1;

片段 C：能编译，但 for x in rx.iter() 会 panic。

rx.iter() 内部对 recv() 的 Err 做了 unwrap()。这里 tx 与 tx2 都被移进了子线程， 两个线程结束后 Sender 全部 drop，通道关闭，iter() 随即 panic：

thread 'main' panicked at ... called `Result::unwrap()` on an `Err` value: RecvError

要优雅结束应写成：

while let Ok(x) = rx.recv() { v.push(x); }
assert_eq!(v.len(), 2);

（注意：v 里的元素顺序不确定，所以只断言长度，不要断言内容顺序。）

片段 D：不能编译。

Rc<Mutex<Vec<i32>>> 里的 Mutex 并不能让 Rc 变成 Send： Rc<T> 有 impl !Send for Rc<T> / impl !Sync for Rc<T> 的负实现，与 T 是什么完全无关。错误是：

error[E0277]: `Rc<Mutex<Vec<i32>>>` cannot be sent between threads safely
   = help: the trait `Send` is not implemented for `Rc<Mutex<Vec<i32>>>`

正确写法是把 Rc 换成 Arc：Arc<Mutex<Vec<i32>>>，此时 Arc<T>: Send 需要 T: Send + Sync， 而 Mutex<Vec<i32>> 满足（Vec<i32>: Send），编译通过且断言成立。

要点解析：这四段覆盖了本章最重要的四个判断维度——"原子操作是否足够"、"锁的粒度是否覆盖了整个不变式"、"迭代器 API 的失败语义"、"包装类型是否改变了 auto trait"。 记住：能编译 ≠ 正确；不能编译 ≠ 你写错了（可能是选错了类型）。

练习 9：用 Condvar 实现有界阻塞队列

难度：★★☆

要求：实现一个泛型 BoundedQueue<T>，内部用 Mutex<VecDeque<T>> + 两个 Condvar（not_empty/not_full）， 提供 push(&self, v: T)（满则阻塞）与 pop(&self) -> T（空则阻塞），容量在 new(cap) 时指定。用 1 个生产者、2 个消费者各跑一遍， 断言所有元素恰好被消费一次。

提示：wait 必须放在 while 里；WaitTimeoutResult 用不到时直接 q = cv.wait(q).unwrap()。

参考答案（先自己写再看）

参考答案（先自己写再看）

use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex};
use std::thread;

pub struct BoundedQueue<T> {
    inner: Mutex<VecDeque<T>>,
    capacity: usize,
    not_empty: Condvar,
    not_full: Condvar,
}

impl<T> BoundedQueue<T> {
    pub fn new(capacity: usize) -> Self {
        assert!(capacity > 0, "容量必须为正");
        Self {
            inner: Mutex::new(VecDeque::with_capacity(capacity)),
            capacity,
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }

    /// 队列满时阻塞，直到有空位
    pub fn push(&self, value: T) {
        let mut q = self.inner.lock().unwrap();
        // 必须 while：wait 允许虚假唤醒，且被唤醒后条件可能又变了
        while q.len() >= self.capacity {
            q = self.not_full.wait(q).unwrap();
        }
        q.push_back(value);
        drop(q); // 先释放锁再通知，避免唤醒者立刻又阻塞在锁上
        self.not_empty.notify_one();
    }

    /// 队列空时阻塞，直到有元素
    pub fn pop(&self) -> T {
        let mut q = self.inner.lock().unwrap();
        while q.is_empty() {
            q = self.not_empty.wait(q).unwrap();
        }
        let v = q.pop_front().expect("while 保证了非空");
        drop(q);
        self.not_full.notify_one();
        v
    }

    pub fn len(&self) -> usize {
        self.inner.lock().unwrap().len()
    }

    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// 非阻塞取出：用于"取一件就走"的场景，返回 None 表示此刻为空
    pub fn try_pop(&self) -> Option<T> {
        let mut q = self.inner.lock().unwrap();
        let v = q.pop_front();
        if v.is_some() {
            drop(q);
            self.not_full.notify_one();
        }
        v
    }
}

fn main() {
    let q = Arc::new(BoundedQueue::new(3));
    let total = 200;

    let producer = {
        let q = Arc::clone(&q);
        thread::spawn(move || {
            for i in 0..total {
                q.push(i);
            }
        })
    };

    let consumers: Vec<_> = (0..2)
        .map(|_| {
            let q = Arc::clone(&q);
            thread::spawn(move || {
                let mut seen = Vec::new();
                // 关键问题：消费者怎么知道"不会再有新元素了"？
                // 队列本身只能告诉你"此刻为空"，无法区分"暂时空"与"永远空"。
                while let Some(v) = q.try_pop() {
                    seen.push(v);
                    if seen.len() >= total / 2 { break; }
                }
                seen
            })
        })
        .collect();
    let _ = (producer, consumers);
    println!("BoundedQueue 结构已完成，但消费者缺少终止条件，见下方完整版本");
}

注意上面 main 的两个问题：try_pop 会忙等浪费 CPU，而且按数量提前 break 会让元素丢失（某个消费者多拿了，另一个就提前退出）。 "何时该停"无法只靠队列判断，需要一条外部信号。下面是完整且可运行的版本： 生产者结束后置一个 AtomicBool，消费者在"不再生产 且 队列已空"时才退出。

use std::collections::VecDeque;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread;

pub struct BoundedQueue<T> {
    inner: Mutex<VecDeque<T>>,
    capacity: usize,
    not_empty: Condvar,
    not_full: Condvar,
}

impl<T> BoundedQueue<T> {
    pub fn new(capacity: usize) -> Self {
        assert!(capacity > 0, "容量必须为正");
        Self {
            inner: Mutex::new(VecDeque::with_capacity(capacity)),
            capacity,
            not_empty: Condvar::new(),
            not_full: Condvar::new(),
        }
    }

    /// 队列满时阻塞，直到有空位
    pub fn push(&self, value: T) {
        let mut q = self.inner.lock().unwrap();
        // 必须 while：wait 允许虚假唤醒，且被唤醒后条件可能又被别人改回不满足
        while q.len() >= self.capacity {
            q = self.not_full.wait(q).unwrap();
        }
        q.push_back(value);
        drop(q); // 先释放锁再通知，避免唤醒者立刻又阻塞在锁上
        self.not_empty.notify_one();
    }

    /// 队列空时阻塞，直到有元素
    pub fn pop(&self) -> T {
        let mut q = self.inner.lock().unwrap();
        while q.is_empty() {
            q = self.not_empty.wait(q).unwrap();
        }
        let v = q.pop_front().expect("while 保证了此刻非空");
        drop(q);
        self.not_full.notify_one();
        v
    }

    /// 非阻塞取出：None 表示此刻为空（不代表以后不会有）
    pub fn try_pop(&self) -> Option<T> {
        let mut q = self.inner.lock().unwrap();
        let v = q.pop_front();
        if v.is_some() {
            drop(q);
            self.not_full.notify_one();
        }
        v
    }

    pub fn is_empty(&self) -> bool {
        self.inner.lock().unwrap().is_empty()
    }
}

fn main() {
    const TOTAL: usize = 200;
    let q = Arc::new(BoundedQueue::new(3));
    // 队列本身无法表达"生产者已结束"，用外部标志补充这个信息
    let producing = Arc::new(AtomicBool::new(true));

    let producer = {
        let q = Arc::clone(&q);
        thread::spawn(move || {
            for i in 0..TOTAL {
                q.push(i); // 队列满时自动阻塞，形成背压
            }
        })
    };

    let consumers: Vec<_> = (0..2)
        .map(|_| {
            let q = Arc::clone(&q);
            let producing = Arc::clone(&producing);
            thread::spawn(move || {
                let mut got = Vec::new();
                loop {
                    match q.try_pop() {
                        Some(v) => got.push(v),
                        // 没有数据：只有"生产者已结束且队列已空"才退出
                        None => {
                            if !producing.load(Ordering::SeqCst) && q.is_empty() {
                                break;
                            }
                            thread::yield_now(); // 让出 CPU，避免纯自旋抢占
                        }
                    }
                }
                got
            })
        })
        .collect();

    producer.join().unwrap(); // 结束时保证所有 push 都已 happens-before
    producing.store(false, Ordering::SeqCst); // 再宣布"不再生产"

    let mut all: Vec<usize> = Vec::new();
    for c in consumers {
        all.extend(c.join().unwrap());
    }
    all.sort_unstable();
    assert_eq!(all.len(), TOTAL, "每个元素必须恰好被消费一次");
    assert_eq!(all, (0..TOTAL).collect::<Vec<_>>());
    println!("共消费 {} 个元素，无丢失无重复", all.len());
}

输出：共消费 200 个元素，无丢失无重复

要点解析：

1. wait 一定要包在 while 里，防虚假唤醒，也防"被唤醒后条件又被别的线程改回不满足"。
2. notify_one 放在释放锁之后，减少"唤醒 → 抢锁 → 又睡"的空转；这不是正确性要求，但能显著降低无效唤醒。
3. 终止条件必须来自队列之外。try_pop() 返回 None 只说明"此刻为空"，无法区分"生产者还在算"和"生产者已经结束"。三种常见做法：外部标志位（本例）、 "毒丸"（发一个特殊值表示结束）、以及只用一个消费者（此时可用 pop() 阻塞等待）。
4. 不要用"按数量分配"的原子计数来替代。像 remaining.fetch_sub(1) 这种"先扣名额再取数据"的写法在多消费者下是错的： fetch_sub 命中 0 的线程会直接退出，而其他线程可能已经多取或少取了元素；如果这些线程随后调用阻塞的 pop()，还会因为生产者已经结束而永久挂死。 名额分配必须与"取走元素"在同一个临界区内完成。
5. 消费者用 try_pop + yield_now 是"折中方案"；更优雅的做法是让 pop 支持超时（配合 Condvar::wait_timeout） 或额外加一个 Condvar 专门用于"生产结束"的广播。真实项目里直接用 crossbeam-channel 的 recv/select! 更省心。

练习 11：用 OnceLock 并发初始化缓存

难度：★★☆

要求：用 static CACHE: OnceLock<HashMap<u32, u32>> 配合 get_or_init，让 8 个线程并发请求同一个 key， 验证昂贵的初始化代码只执行一次（用一个 static INIT_COUNT: AtomicUsize 计数，断言为 1）。

提示：get_or_init 是原子的"检查并初始化"，天然避免练习 8 片段 B 那类竞态。

参考答案（先自己写再看）

参考答案（先自己写再看）

use std::collections::HashMap;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::OnceLock;
use std::thread;
use std::time::Duration;

static CACHE: OnceLock<HashMap<u32, u32>> = OnceLock::new();
static INIT_COUNT: AtomicUsize = AtomicUsize::new(0);

/// 模拟昂贵初始化：无论多少线程同时调用，闭包体只执行一次
fn expensive_table() -> &'static HashMap<u32, u32> {
    CACHE.get_or_init(|| {
        INIT_COUNT.fetch_add(1, Ordering::SeqCst);
        thread::sleep(Duration::from_millis(50)); // 放大竞争窗口
        let mut m = HashMap::new();
        for k in 0..100 {
            m.insert(k, k * k);
        }
        m
    })
}

fn main() {
    let handles: Vec<_> = (0..8)
        .map(|id| {
            thread::spawn(move || {
                // 所有线程请求同一个 key
                let v = expensive_table().get(&42).copied();
                (id, v)
            })
        })
        .collect();

    for h in handles {
        let (id, v) = h.join().unwrap();
        assert_eq!(v, Some(42 * 42), "线程 {id} 读到了错误的表");
    }

    assert_eq!(INIT_COUNT.load(Ordering::SeqCst), 1, "初始化必须只执行一次");
    println!("8 个线程全部拿到 1764，初始化次数 = 1");
}

输出：8 个线程全部拿到 1764，初始化次数 = 1

要点解析：get_or_init 的内部实现保证了"竞争时只有一个线程执行初始化闭包，其余线程阻塞等待结果"。 这正好解决了练习 8 片段 B 那种 check-then-act 竞态——而且没有把整张表的读写暴露给调用方去加锁，访问路径零开销（初始化后只是一次 load）。 如果要缓存可变的全局表，用 static CACHE: LazyLock<Mutex<HashMap<..>>>（1.80+）或者 OnceLock<Mutex<..>>。 INIT_COUNT 用 SeqCst 是为了让断言在任何平台上都稳，统计类计数其实 Relaxed 就够。

练习 12：用 rayon 做数据并行

难度：★★★

要求：cargo add rayon 后，对一个 Vec<u64>（100 万个元素）分别用串行 iter() 和 par_iter() 计算"偶数之和"， 断言结果相同，并用 std::time::Instant 打印两者耗时。

提示：use rayon::prelude::*; 之后 iter → par_iter 即可；本地机器核数少时并行不一定更快，请如实报告观测结果并解释原因。

参考答案（先自己写再看）

参考答案（先自己写再看）

cargo new even_sum --bin
cd even_sum
cargo add rayon        # 会写入 rayon = "1"

use rayon::prelude::*;
use std::time::Instant;

fn main() {
    let data: Vec<u64> = (1..=1_000_000).collect();

    let t0 = Instant::now();
    let serial: u64 = data.iter().filter(|x| *x % 2 == 0).sum();
    let serial_time = t0.elapsed();

    let t1 = Instant::now();
    // 只是把 iter 换成 par_iter，rayon 负责切分与工作窃取
    let parallel: u64 = data.par_iter().filter(|x| *x % 2 == 0).sum();
    let parallel_time = t1.elapsed();

    assert_eq!(serial, parallel, "并行结果必须与串行完全一致");
    println!("串行 {serial} 用时 {serial_time:?}");
    println!("并行 {parallel} 用时 {parallel_time:?}");
    println!("期望值（手工推导）= 250000500000");
}

预期输出（数值确定，耗时因机器而异）：
串行 250000500000 用时 约 1.5ms
并行 250000500000 用时 约 0.5ms
期望值（手工推导）= 250000500000

手工推导： 1..=1_000_000 中的偶数之和 = 2 × (1 + 2 + ... + 500_000) = 2 × 500_000 × 500_001 / 2 = 500_000 × 500_001 = 250_000_500_000。

要点解析：par_iter 能编译通过本身就说明"这个闭包是线程安全的"——rayon 的 trait 约束（Send + Sync）会把不安全的用法挡在编译期。 这类"纯映射/归约"的计算是 rayon 的最佳场景；反过来，如果闭包里需要访问共享可变状态（例如往同一个 Vec 里 push）， 要么改成 fold/reduce 的归约形式，要么回到 Mutex。注意：数据量小时并行反而更慢（线程调度开销 > 计算收益），上面的耗时数字只是量级示例， 请以你自己机器的实测为准并解释差异。

本章小结 / 自测清单

• 我能用一句话说清"并发"与"并行"的区别，并说出绿色线程被 std 放弃的原因。
• 我能列出"无畏并发"排除了什么、没有排除什么（至少说出死锁、竞态条件、内存泄漏三项）。
• 我知道 spawn 的三个约束（FnOnce + Send + 'static）分别对应哪个设计动机。
• 我能在需要借用局部数据并行时立刻想到 thread::scope，而不是 move + 克隆。
• 我能解释 join() 为什么返回 Result，并会 downcast_ref 取出 panic 负载。
• 我知道 mpsc 的 Sender 可克隆、Receiver 不可克隆，并会用 drop(tx) 优雅关闭。
• 我能说出 sync_channel(n) 的背压语义，以及 n == 0 的会合通道用途。
• 我能解释为什么 Arc<Receiver> 编译不过，并写出两种多消费者替代方案。
• 我能说清 Rust Mutex<T> 与 Java synchronized 的根本区别（锁在类型里 vs 锁代码块）。
• 我知道 MutexGuard 是 RAII 解锁、!Send、可能中毒，并选好了项目的 unwrap 策略。
• 我能说出至少三条避免死锁的纪律，并能读懂 try_lock 的取舍。
• 我能背出 Ordering 五个变体的用途，并知道"默认先 SeqCst"的理由。
• 我能说出"什么时候用原子而不是 Mutex"，以及原子为什么保护不了复合不变式。
• 我能默写 T: Sync ⟺ &T: Send，并解释 Rc、RefCell、MutexGuard、裸指针各缺哪个。
• 我知道 unsafe impl Send/Sync 要写 // SAFETY: 论证，且只在最内层类型上使用。
• 我能区分数据竞争与竞态条件，并能改写"读-改-写跨两个临界区"的代码。
• 我会用 cargo test -- --test-threads=1、--nocapture 以及反复运行来排查并发问题。
• 我能说出"CPU 密集用线程/rayon、I/O 密集用 async"的判断标准。