使用par_iter处理大数组时无法更新多个可变数组的问题

问题根源 #

你用par_iter并行遍历数组时，直接在闭包里修改errors和assets_record这两个可变Vec。Rust的并行迭代闭包要求是Fn类型（不可变捕获），因为多个线程同时无同步地修改同一个变量会引发数据竞争，所有权系统直接阻止了这种不安全操作，所以报E0596错误。

解决方案 #

方案1：用线程安全容器（Arc）共享可变状态 #

通过Arc<Mutex>包装需要共享的Vec，让多个线程可以安全地获取可变引用：

use std::sync::{Arc, Mutex};
use rayon::prelude::*;

fn par_calculate_depreciation_for_assets(assets_to_depreciate: &Vec<Depreciation>) -> Result<(), Ferror> {
    let errors = Arc::new(Mutex::new(Vec::<String>::new()));
    let assets_record = Arc::new(Mutex::new(Vec::<AssetRecord>::new()));

    assets_to_depreciate.par_iter().for_each(|item| {
        let result = calculate_monthly_depreciation(item);
        
        if result.is_err() {
            // 锁定Mutex后再修改errors
            let mut err_list = errors.lock().unwrap();
            err_list.push(item.asset_code.clone());
        } else {
            let (asset, depreciated_asset) = result.unwrap();
            // 锁定Mutex后再修改assets_record
            let mut record_list = assets_record.lock().unwrap();
            build_csv_struct(item, &mut record_list, asset);
        }
    });
    
    // 提取内部的Vec
    let assets_record = assets_record.lock().unwrap().clone();
    let errors = errors.lock().unwrap().clone();
    
    append_to_csv(assets_record);
    dbg!("Errors {:?} unprocessed", errors);
    Ok(())
}

方案2：用Rayon的map+partition拆分结果（更高效） #

这种方式避免锁竞争，先让每个线程独立处理元素，返回结果后再统一拆分成功/失败集合：

use rayon::prelude::*;

// 定义一个枚举，存储每个元素的处理结果
enum ProcessingResult {
    Success(AssetRecord),
    Error(String),
}

fn par_calculate_depreciation_for_assets(assets_to_depreciate: &Vec<Depreciation>) -> Result<(), Ferror> {
    let results: Vec<ProcessingResult> = assets_to_depreciate.par_iter()
        .map(|item| {
            match calculate_monthly_depreciation(item) {
                Ok((asset, _depreciated_asset)) => {
                    let mut record = AssetRecord::default(); // 假设build_csv_struct构建AssetRecord
                    build_csv_struct(item, &mut record, asset);
                    ProcessingResult::Success(record)
                }
                Err(_) => ProcessingResult::Error(item.asset_code.clone()),
            }
        })
        .collect();
    
    // 拆分成功和失败的结果
    let (success_records, error_codes): (Vec<ProcessingResult>, Vec<ProcessingResult>) = results.into_iter()
        .partition(|res| matches!(res, ProcessingResult::Success(_)));
    
    // 转换成功的结果
    let assets_record: Vec<AssetRecord> = success_records.into_iter()
        .filter_map(|res| match res {
            ProcessingResult::Success(r) => Some(r),
            _ => None,
        })
        .collect();
    
    // 转换错误的结果
    let errors: Vec<String> = error_codes.into_iter()
        .filter_map(|res| match res {
            ProcessingResult::Error(e) => Some(e),
            _ => None,
        })
        .collect();
    
    append_to_csv(assets_record);
    dbg!("Errors {:?} unprocessed", errors);
    Ok(())
}

方案对比 #

• 方案1：适合需要实时收集结果的场景，但频繁加锁解锁会带来性能开销，数据量大时可能变慢。
• 方案2：更高效，因为每个线程独立处理，最后合并结果，没有锁竞争，推荐在数据量大时使用。

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