如何在Rust中使用Rayon的并行fold方法创建HashMap

报错原因 #

rayon的并行迭代器fold方法设计和标准库原生迭代器不同：

• 标准库fold第一个参数是固定初始值，单线程迭代只需要初始化一次累加器
• rayon并行fold第一个参数要求是无参闭包，每个参与计算的线程会独立调用这个闭包生成自己的本地累加器（也就是独立的HashMap实例），从根源避免多线程同时修改同一个哈希表的竞态问题，这就是第一个类型报错的原因。

同时并行fold执行完后只会返回每个线程生成的中间HashMap集合，还需要额外调用reduce方法把所有中间哈希表合并成最终的总表，这是第二个类型不匹配报错的原因。

修复后的完整代码 #

use rayon::prelude::*; // 1.5.1
use std::collections::HashMap;

fn main() {
    let nums = [1, 2, 1, 2, 1, 2];
    let result: HashMap<i32, i32> =
        nums.par_iter()
            .filter(|x| *x % 2 == 0)
            // 每个线程初始化自己的本地HashMap
            .fold(|| HashMap::new(), |mut acc, x| {
                *acc.entry(*x).or_insert(0) += 1;
                acc
            })
            // 合并所有线程的本地HashMap到总表
            .reduce(|| HashMap::new(), |mut total, local_map| {
                for (k, v) in local_map {
                    *total.entry(k).or_insert(0) += v;
                }
                total
            });

    println!("{:?}", result);
}

运行逻辑说明 #

1. rayon自动将输入数组拆分到多个CPU核心的线程上执行
2. 每个线程处理自己分到的部分数据，把计数写入自己本地的HashMap，全程无跨线程共享可变状态
3. 所有线程计算完成后，逐步合并各个本地HashMap的计数，得到最终的全局统计结果

内容的提问来源于stack exchange，提问作者leviathan