Rayon 并行迭代器聚合

示例说明

rayon 的并行迭代器把集合切成工作窃取的任务池：每个工作线程先本地 fold 出局部结果，最后 reduce 合并——全程无锁无原子竞争，合并阶段的成本远低于共享计数器的争用。par_chunks 更是显式声明分块边界：每块的求和完全独立，块间零共享状态。

技术要点

• fold 与 reduce 组合实现无锁聚合
• 工作窃取自动负载均衡
• par_chunks 显式分块隔离状态

代码解读

word_frequencies 按行并行分词，fold 每线程构建局部词频表，reduce 把各表逐词条累加进结果。checksum_parallel 以 256KB 为块并行求和再汇总。最后一个查询演示更简单的 map/max 形态——并行版本与串行迭代器 API 形状一致，心智负担最小。

使用建议

数据量小别并行——线程池调度开销可能超过收益。嵌套并行要避免死锁。CPU 核数默认即池大小，容器配额环境下显式设定。

代码示例

use rayon::prelude::*;

fn word_frequencies(lines: &[String]) -> std::collections::HashMap<String, usize> {
    lines
        .par_iter()
        .flat_map(|line| line.split_whitespace())
        .fold(
            || std::collections::HashMap::new(),
            |mut acc, word| {
                *acc.entry(word.to_lowercase()).or_insert(0) += 1;
                acc
            },
        )
        .reduce(
            |mut a, b| {
                for (word, count) in b {
                    *a.entry(word).or_insert(0) += count;
                }
                a
            },
        )
}

fn checksum_parallel(bytes: &[u8]) -> u64 {
    bytes
        .par_chunks(1024 * 256)
        .map(|chunk| {
            // 每块独立校验：避免跨块状态
            chunk.iter().map(|&b| b as u64).sum::<u64>()
        )
        .sum()
}

fn main() {
    let corpus: Vec<String> = vec![
        "the quick brown fox".into(),
        "the lazy dog sleeps".into(),
        "brown fox jumps high".into(),
    ];

    let freqs = word_frequencies(&corpus);
    for (word, count) in freqs.iter().take(3) {
        println!("{word}: {count}");
    }

    let data = vec![7u8; 1024 * 1024];
    println!("checksum: {}", checksum_parallel(&data));

    let max_len = corpus
        .par_iter()
        .map(|line| line.len())
        .max()
        .unwrap_or(0);
    println!("longest line: {max_len}");
}