用 rayon 做数据并行
你会学到什么
- 数据并行（data parallelism）：同一份运算，分给多个核心同时处理一批数据。
- rayon 的核心招牌——把
iter()换成par_iter()就能并行，几乎零改动。 - 工作窃取（work-stealing）线程池替你完成切片、分发与负载均衡。
- 为什么并行结果依旧正确：运算是纯函数、归约满足结合律。
- 并行排序
par_sort/par_sort_unstable。 - 什么时候并行划算（CPU 密集、数据量大），什么时候反而更慢（数据太小、开销盖过收益）。
这一章同时建立在迭代器和多线程之上：你已经会写 map/filter/sum 的链式风格，也知道线程能并发跑——rayon 把两者揉在一起，让你「用迭代器的写法，拿到多线程的速度」。
核心招牌：iter() -> par_iter() 即可并行
先看顺序版本，一个再普通不过的迭代器链：
let total: u64 = nums.iter().map(|&n| n * n).sum();改成并行，只动一个词：
use rayon::prelude::*;
let total: u64 = nums.par_iter().map(|&n| n * n).sum();
// ^^^^^^^^ 只把 iter 换成 par_itermap、filter、sum、count、reduce 这些方法名一模一样，链式写法一模一样。背后却天差地别：par_iter 返回的是并行迭代器，rayon 会把 nums 切成若干块，丢进一个全局的工作窃取线程池——空闲的线程会主动「偷」别人还没做完的活，于是各核心都不闲着，调度完全不用你操心，也不必手写 thread::spawn 或 channel。
为什么结果还是对的
并行改变了计算顺序，结果却和顺序版逐位相同。原因有两条，缺一不可：
- 纯函数：
map里的闭包|&n| n * n不读写共享可变状态，每个元素独立算，谁先谁后都一样。 - 结合律：
sum、count、reduce这类归约满足结合律，(a+b)+c == a+(b+c)。rayon 把数组拆成几段、各自求局部和、再合并总和——和从头加到尾完全等价。
所以测试里可以放心断言精确值：平方和、素数个数、排序结果都是确定的，不存在「并行所以结果会抖」。这也是判断「一段迭代器能不能安全并行」的黄金标准：闭包是纯的、归约可结合，就能换 par_iter。
反例：如果归约不满足结合律（比如浮点求和对舍入顺序敏感），并行结果可能和顺序版有极小差异——这时要么接受，要么换确定性的归约策略。本章的整数运算没有这个问题。
文件结构
示例按职责拆成两个模块，方便对照阅读：
50_rayon/
├── Cargo.toml # 声明 rayon = "1"
└── src/
├── lib.rs # mod 声明 + pub use 重导出 + 库级文档
├── parallel_map.rs # 并行 map/filter 与归约：平方和、素数计数
├── sorting.rs # 并行排序：par_sort_unstable
└── main.rs # 薄壳：调用库函数、打印结果、粗略对比串/并行耗时lib.rs 把函数 pub use 出来，main.rs 因此能直接 use rt_50_rayon::par_sum_of_squares;，感知不到底层用了 rayon。每个模块都带就近的 #[cfg(test)] 单元测试。
运行代码
cd examples
cargo run -p rt_50_rayon
cargo test -p rt_50_rayon代码讲解
并行归约（parallel_map.rs）
par_sum_of_squares 就是开头那个例子：par_iter().map(...).sum()。测试里既对比了闭式公式 n(n+1)(2n+1)/6，也和顺序迭代器逐位比较，确认并行不影响结果。
par_count_primes 展示了并行真正擅长的场景——CPU 密集且各任务独立：
pub fn par_count_primes(upto: u64) -> usize {
(0..upto).into_par_iter().filter(|&n| is_prime(n)).count()
}(0..upto) 是个区间，into_par_iter() 把它变成并行迭代器（区间没有 par_iter 的引用版本，按值消费用 into_par_iter）。每个数独立做一次试除判素，彼此不共享可变状态，rayon 把区间切片分给各核——这正是数据并行最理想的形态。filter + count 是可结合归约，所以 [0, 100) 永远数出 25 个素数，[0, 1000) 永远是 168 个。
并行排序（sorting.rs）
排序同样「换名即并行」：
pub fn par_sorted(mut v: Vec<i64>) -> Vec<i64> {
v.par_sort_unstable(); // 顺序版是 v.sort_unstable()
v
}par_sort_unstable 在切片上就地并行排序，内部用并行归并。结果与顺序排序逐位相同，测试里直接和 let mut s = v.clone(); s.sort(); 对比。
「unstable（不稳定）」指相等元素的相对顺序可能被打乱。对 i64 这种纯值没有影响，而且不稳定版通常更快、内存更省。只有当元素是「带 key 的结构体、且要保持同 key 的原始先后」时，才需要换成稳定的 par_sort。
什么时候并行划算
并行不是免费的：切片、分发、跨线程合并都有开销。一个朴素但管用的判断：
- ✅ 值得并行：CPU 密集（每个元素的运算不便宜，如判素、哈希、图像像素处理）、且数据量大。开销被摊薄，多核收益明显。
- ❌ 不值得并行：数据很小（几十、几百个元素），或每个元素的运算极轻（如简单加法）。此时调度开销可能比计算本身还大，
par_iter反而比iter慢。 - ⚠️ 并行帮不上忙：瓶颈在 I/O 或锁等待，而非 CPU。这类「等待型」并发更适合用线程 + channel 或 async，而不是 rayon 的数据并行。
main.rs 里对 200 万以内的素数计数做了一次粗略的串/并行计时对比，你能直观看到差距——但注意它不做断言，因为加速比取决于机器核数和当时负载，不是确定值。
常见错误
忘记导入 prelude：
// ❌ 没有 use rayon::prelude::*;
let s: u64 = nums.par_iter().map(|&n| n * n).sum();
// ^^^^^^^^ 方法不存在par_iter、into_par_iter、par_sort 等都挂在 rayon 的扩展 trait 上，必须 use rayon::prelude::*; 才可见——和上一章 itertools 同一个套路：use 一个 prelude，已有类型凭空多出新方法。
另一个坑是「以为换了 par_iter 就一定更快」：小数据上它可能更慢，务必结合上一节判断场景，必要时实测。
练习
- 把
par_count_primes改成返回所有素数的Vec（提示：filter(...).collect()，并行collect会自动按原顺序拼接）。 - 用
par_iter().map(...).max()并行求一组数据里「某个开销不小的指标」的最大值，并和顺序版对拍。 - 用
reduce自己实现并行求和：par_iter().copied().reduce(|| 0, |a, b| a + b)，体会「单位元 + 结合的合并函数」这套接口。
小结
rayon 把数据并行变成了迭代器的「同名替换」：iter() -> par_iter()、sort() -> par_sort()，工作窃取线程池负责所有调度。只要运算是纯函数、归约可结合，结果就和顺序版完全一致，可以断言精确值。记住「CPU 密集 + 大数据才划算」这条原则，比记住任何单个方法都重要。
下一步
到这里，「引用第三方包」这一组就完整收尾了：从声明依赖、use 进来，到把第三方能力封装进自己的库模块，这套流程你已经走了一遍又一遍。接下来进入 Web 与数据库实战——先看怎样用 axum 写一个 Web 服务，再到 SQLite 做持久化。
完整示例代码
下面是 examples/50_rayon/ 的完整源码。无需 clone 仓库，直接在页面上阅读、复制、对照运行。
examples/50_rayon/src/main.rs
//! 瘦入口：调用库里的并行函数，打印结果，并粗略对比一次串行/并行耗时。
//!
//! 注意：耗时只用于直观感受，不做任何断言——真实加速比取决于机器核数、
//! 数据规模与系统负载。
use std::time::Instant;
use rt_50_rayon::{par_count_primes, par_sorted, par_sum_of_squares};
fn main() {
println!("=== rayon 数据并行 ===\n");
// 1. 并行归约：平方和
let nums: Vec<u64> = (1..=1_000).collect();
println!("[平方和] 1..=1000 的平方和 = {}", par_sum_of_squares(&nums));
// 2. 并行排序
let v = vec![5, 3, 8, 1, 9, 2, 7, 4, 6, 0];
let sorted = par_sorted(v.clone());
println!("[排序] {v:?} -> {sorted:?}");
// 3. CPU 密集型并行：素数计数（顺手做一次粗略计时对比）
let upto = 2_000_000;
let t0 = Instant::now();
let seq = (0..upto)
.filter(|&n| {
if n < 2 {
return false;
}
let mut d = 2;
while d * d <= n {
if n % d == 0 {
return false;
}
d += 1;
}
true
})
.count();
let seq_dur = t0.elapsed();
let t1 = Instant::now();
let par = par_count_primes(upto);
let par_dur = t1.elapsed();
println!("\n[素数] [0, {upto}) 内素数个数 = {par}");
println!(" 顺序耗时 ≈ {seq_dur:?}");
println!(" 并行耗时 ≈ {par_dur:?}");
println!(" （两者结果一致：{}）", seq == par);
println!("\n提示：数据越大、单个任务越重，并行收益越明显；数据很小时反受调度开销拖累。");
}examples/50_rayon/src/lib.rs
//! 用 [`rayon`](https://docs.rs/rayon) 做数据并行（data parallelism）。
//!
//! rayon 最迷人的一点：把迭代器链里的 `iter()` 换成 `par_iter()`，几乎零改动
//! 就能利用多核。它内部维护一个工作窃取（work-stealing）线程池，自动把任务
//! 切片、分发、做负载均衡，你不必手写 `thread::spawn` 或管理通道。
//!
//! 之所以「换个名字就行」还能保证结果正确，是因为这些操作是**纯函数**、
//! 归约满足**结合律**：拆开分别算、再把局部结果合并，和顺序计算等价。
//!
//! 本章覆盖：
//! - [`par_sum_of_squares`]：`par_iter().map().sum()` 的并行归约；
//! - [`par_count_primes`]：`into_par_iter().filter().count()` 的 CPU 密集型并行；
//! - [`par_sorted`]：`par_sort_unstable` 的就地并行排序。
//!
//! 代码按职责拆分为两个模块：
//! - [`parallel_map`]：并行 `map` / `filter` 与归约；
//! - [`sorting`]：并行排序。
pub mod parallel_map;
pub mod sorting;
pub use parallel_map::{par_count_primes, par_sum_of_squares};
pub use sorting::par_sorted;examples/50_rayon/src/parallel_map.rs
//! 并行的 `map` / `filter` / 归约：把迭代器链里的 `iter()` 换成 `par_iter()`。
//!
//! rayon 的核心承诺是「几乎零改动」：原本写 `nums.iter().map(...).sum()`，
//! 只需把 `iter()` 改成 `par_iter()`，rayon 的工作窃取（work-stealing）线程池
//! 就会把任务切片、分发到多个核心上跑。结果依旧正确，因为这些运算是
//! 纯函数 + 满足结合律（associative）的归约——拆开求和再合并，和顺序求和等价。
use rayon::prelude::*;
/// 并行计算「每个元素平方之和」。
///
/// 顺序版本是 `nums.iter().map(|&n| n * n).sum()`，这里只把 `iter` 换成
/// `par_iter`。`map` 是纯函数、`sum` 满足结合律，所以无论 rayon 怎样切分、
/// 用多少线程，结果都和顺序求和完全一致——可以在测试里断言精确值。
///
/// ```
/// use rt_50_rayon::par_sum_of_squares;
/// assert_eq!(par_sum_of_squares(&[1, 2, 3]), 14);
/// ```
pub fn par_sum_of_squares(nums: &[u64]) -> u64 {
nums.par_iter().map(|&n| n * n).sum()
}
/// 判断 `n` 是否为素数（朴素试除法）。
///
/// 这是一个纯 CPU 的小任务，单个不算重，但对每个数都做一遍、数量一大，
/// 总开销就足以让并行化获益。
fn is_prime(n: u64) -> bool {
if n < 2 {
return false;
}
if n % 2 == 0 {
return n == 2;
}
let mut d = 3;
while d * d <= n {
if n % d == 0 {
return false;
}
d += 2;
}
true
}
/// 并行统计 `[0, upto)` 区间内素数的个数。
///
/// `(0..upto)` 是一个区间，`into_par_iter()` 把它变成并行迭代器；每个数独立
/// 做一次 `is_prime` 判断，互不依赖——这正是数据并行（data parallelism）最
/// 理想的场景：CPU 密集、任务之间没有共享可变状态。`filter` + `count` 是
/// 可结合的归约，所以并行结果与顺序结果相同。
///
/// 例如 `[0, 100)` 内有 25 个素数。
///
/// ```
/// use rt_50_rayon::par_count_primes;
/// assert_eq!(par_count_primes(100), 25);
/// ```
pub fn par_count_primes(upto: u64) -> usize {
(0..upto).into_par_iter().filter(|&n| is_prime(n)).count()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn sum_of_squares_matches_sequential() {
let nums: Vec<u64> = (1..=1000).collect();
// 顺序参考值：1²+2²+…+n² = n(n+1)(2n+1)/6
let n = 1000u64;
let expected = n * (n + 1) * (2 * n + 1) / 6;
assert_eq!(par_sum_of_squares(&nums), expected);
// 也与顺序迭代器逐位比较，确认并行结果完全一致。
let seq: u64 = nums.iter().map(|&x| x * x).sum();
assert_eq!(par_sum_of_squares(&nums), seq);
}
#[test]
fn sum_of_squares_small_and_empty() {
assert_eq!(par_sum_of_squares(&[]), 0);
assert_eq!(par_sum_of_squares(&[1, 2, 3]), 14);
}
#[test]
fn count_primes_known_values() {
// 经典基准：100 以内 25 个、1000 以内 168 个素数。
assert_eq!(par_count_primes(100), 25);
assert_eq!(par_count_primes(1000), 168);
}
#[test]
fn count_primes_matches_sequential() {
let upto = 5000;
let seq = (0..upto).filter(|&n| is_prime(n)).count();
assert_eq!(par_count_primes(upto), seq);
}
#[test]
fn is_prime_edge_cases() {
assert!(!is_prime(0));
assert!(!is_prime(1));
assert!(is_prime(2));
assert!(is_prime(3));
assert!(!is_prime(9));
assert!(is_prime(97));
}
}examples/50_rayon/src/sorting.rs
//! 并行排序：`par_sort` / `par_sort_unstable` 是标准库 `sort` 的就地并行版。
//!
//! 和并行迭代器一样，这里也是「换个名字就能并行」：把 `v.sort()` 换成
//! `v.par_sort()`，rayon 会在内部用并行归并把切片排好。排序结果与顺序排序
//! **逐位相同**，因此可以在测试里和 `Vec::sort` 的输出精确对比。
use rayon::prelude::*;
/// 并行排序：消费一个 `Vec`，返回升序排好的新 `Vec`。
///
/// 用的是 `par_sort_unstable`——「不稳定」指相等元素的相对顺序可能改变，
/// 但对 `i64` 这种没有「附带数据」的值类型而言，最终序列与稳定排序完全一致，
/// 而且不稳定版通常更快、内存占用更小。若元素是「带 key 的结构体」且要保持
/// 同 key 的原始先后，则应改用 `par_sort`（稳定版）。
///
/// 无论是否并行，排序的输出是确定的，所以结果等于 `let mut s = v; s.sort();`。
///
/// ```
/// use rt_50_rayon::par_sorted;
/// assert_eq!(par_sorted(vec![3, 1, 2]), vec![1, 2, 3]);
/// ```
pub fn par_sorted(mut v: Vec<i64>) -> Vec<i64> {
v.par_sort_unstable();
v
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn sorts_a_small_vec() {
assert_eq!(par_sorted(vec![3, 1, 2, -5, 0]), vec![-5, 0, 1, 2, 3]);
}
#[test]
fn matches_sequential_sort_on_large_input() {
// 用一个确定的伪随机序列，避免引入额外依赖，同时保证测试可复现。
let mut state: i64 = 1;
let data: Vec<i64> = (0..10_000)
.map(|_| {
// 线性同余发生器（LCG），纯确定性。
state = state.wrapping_mul(6364136223846793005).wrapping_add(1);
state
})
.collect();
let mut expected = data.clone();
expected.sort();
assert_eq!(par_sorted(data), expected);
}
#[test]
fn handles_empty_and_single() {
assert_eq!(par_sorted(vec![]), Vec::<i64>::new());
assert_eq!(par_sorted(vec![42]), vec![42]);
}
}examples/50_rayon/Cargo.toml
[package]
name = "rt_50_rayon"
version.workspace = true
edition.workspace = true
publish.workspace = true
[dependencies]
rayon = "1"