如何用Rayon的par_split处理字符串并收集结果到Vector？

我需要将文件内容读取到第14行定义的字符串messages中，该文件包含多个以数字开头的块。读取完成后，块之间以换行符分隔，块内每行以__SEP__分隔。希望使用Rayon的par_split()方法并行处理messages中的每个块，并将每个块生成的vec_local（第53行定义）收集到向量vec_final中。

现有代码 #

fn starts_with_digit_or_at_sign(inp: &str) -> bool {
    let mut at_sign_found = false;
    if inp.len() > 0 {
        let ch = inp.chars().next().unwrap();
        if ch.is_numeric() || ch == '@' {
            return true;
        }
    }
    return false;
}
fn main() {
    let filepath = "inp.log";
    let data = std::fs::read_to_string(filepath).expect("file not found!");
    let mut messages: String = String::from("");
    let separator_char = '\n';
    let separator: String = String::from("__SEP__");
    let mut found_first_message = false;
    let mut start_of_new_msg = false;
    let mut line_num = 0;
    for line in data.lines() {
        line_num += 1;
        if line.len() > 0 {
            if starts_with_digit_or_at_sign(line) {
                start_of_new_msg = true;
                if !found_first_message {
                    found_first_message = true;
                } else {
                    messages.push(separator_char);
                }
            }
            if found_first_message {
                if !start_of_new_msg {
                    messages.push_str(&separator);
                }
                messages.push_str(line);
                if start_of_new_msg {
                    start_of_new_msg = false;
                    let mut tmp = String::from("Lnumber ");
                    tmp.push_str(&line_num.to_string());
                    messages.push_str(&separator);
                    messages.push_str(&tmp);
                }
            }
        }
    }
    messages.par_split(separator_char).for_each(|l| {
        println!(
            "line: '{}' len: {}, {}",
            l,
            l.len(),
            rayon::current_num_threads()
        );
        let vec_local: Vec<i32> = vec![l.len() as i32];
    }); // <-- line 54
}

代码运行输出 #

line: '1__SEP__Lnumber 1__SEP__a__SEP__b__SEP__c' len: 41, 8
line: '3__SEP__Lnumber 9__SEP__g__SEP__h__SEP__i' len: 41, 8
line: '2__SEP__Lnumber 5__SEP__d__SEP__e__SEP__f' len: 41, 8
line: '4__SEP__Lnumber 13__SEP__j__SEP__k__SEP__l' len: 42, 8

输入文件inp.log内容 #

1
a
b
c
2
d
e
f
3
g
h
i
4
j
k
l

实现指引与修改方案 #

要实现收集vec_local到vec_final的需求，只需对代码的并行处理部分做以下核心修改：

1. 替换for_each为map+collect #

for_each仅用于执行打印这类副作用操作，无法返回结果并收集。改用map在每个块的闭包中返回vec_local，再通过collect()将所有结果汇总到vec_final中。

2. 明确变量类型 #

vec_final的类型为Vec<Vec<i32>>，因为每个块返回一个单元素Vec<i32>，最终会形成嵌套向量。

修改后的并行处理代码片段 #

// 替换原有的for_each块
let vec_final: Vec<Vec<i32>> = messages.par_split(separator_char)
    .map(|l| {
        println!(
            "line: '{}' len: {}, {}",
            l,
            l.len(),
            rayon::current_num_threads()
        );
        vec![l.len() as i32] // 返回当前块生成的vec_local
    })
    .collect(); // 收集所有块的结果到vec_final

// 可选：打印验证最终结果
println!("vec_final: {:?}", vec_final);

完整优化后代码 #

use rayon::prelude::*; // 必须引入Rayon的prelude才能使用par_split

fn starts_with_digit_or_at_sign(inp: &str) -> bool {
    !inp.is_empty() && {
        let ch = inp.chars().next().unwrap();
        ch.is_numeric() || ch == '@'
    }
}

fn main() {
    let filepath = "inp.log";
    let data = std::fs::read_to_string(filepath).expect("file not found!");
    let mut messages = String::new();
    let separator_char = '\n';
    let separator = String::from("__SEP__");
    let mut found_first_message = false;
    let mut start_of_new_msg = false;
    let mut line_num = 0;

    for line in data.lines() {
        line_num += 1;
        if line.is_empty() {
            continue;
        }

        if starts_with_digit_or_at_sign(line) {
            start_of_new_msg = true;
            if !found_first_message {
                found_first_message = true;
            } else {
                messages.push(separator_char);
            }
        }

        if found_first_message {
            if !start_of_new_msg {
                messages.push_str(&separator);
            }
            messages.push_str(line);
            if start_of_new_msg {
                start_of_new_msg = false;
                let tmp = format!("Lnumber {}", line_num);
                messages.push_str(&separator);
                messages.push_str(&tmp);
            }
        }
    }

    let vec_final: Vec<Vec<i32>> = messages.par_split(separator_char)
        .map(|l| {
            println!(
                "line: '{}' len: {}, {}",
                l,
                l.len(),
                rayon::current_num_threads()
            );
            vec![l.len() as i32]
        })
        .collect();

    println!("最终收集的向量: {:?}", vec_final);
}

额外说明 #

• 确保在Cargo.toml中添加Rayon依赖：rayon = "1.8"
• 优化后的代码最终会输出类似：最终收集的向量: [[41], [41], [41], [42]]

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