基于Rust nom7实现带容量限制的流式多行文本解析器

前置依赖 #

先在Cargo.toml中添加nom 7依赖：

[dependencies]
nom = "7.0"

完整实现代码 #

use nom::{
    character::streaming::line_ending,
    error::{Error, ErrorKind},
    Err, IResult,
};
use std::fs::File;
use std::io::{Read, Write};
use std::str;

/// 从输入切片中切割出一行内容（不含行结束符），流式模式
fn parse_line(input: &[u8]) -> IResult<&[u8], &[u8]> {
    let (remaining, line_content) = nom::bytes::streaming::take_until("\n")(input)?;
    let (remaining, _) = line_ending(remaining)?;
    Ok((remaining, line_content))
}

fn main() -> std::io::Result<()> {
    // 打开输入文件和两个输出文件
    let mut input_file = File::open("/words.txt")?;
    let mut valid_file = File::create("valid_ascii.txt")?;
    let mut invalid_file = File::create("invalid_content.txt")?;

    // 1000字节固定缓冲区，对应1000ASCII字符的容量上限
    const BUF_SIZE: usize = 1000;
    let mut buf = vec![0u8; BUF_SIZE];
    // 记录当前缓冲区已填充的有效数据长度
    let mut filled = 0;

    loop {
        // 往缓冲区的剩余空间读入新数据
        let read_count = input_file.read(&mut buf[filled..])?;
        if read_count == 0 && filled == 0 {
            // 读到文件末尾，无剩余数据，退出循环
            break;
        }
        filled += read_count;

        // 循环解析缓冲区中的完整行
        loop {
            match parse_line(&buf[..filled]) {
                Ok((remaining, line)) => {
                    // 校验行是否为不含控制字符的有效ASCII
                    let is_valid = line.iter().all(|&b| b.is_ascii() && !b.is_ascii_control());
                    if is_valid && !line.is_empty() {
                        writeln!(valid_file, "A{}> {}", line.len(), str::from_utf8(line).unwrap())?;
                    } else {
                        writeln!(invalid_file, "C> {}", String::from_utf8_lossy(line))?;
                    }
                    // 计算剩余未解析数据的长度，将剩余数据移到缓冲区头部
                    let remaining_len = remaining.len();
                    buf.copy_within(filled - remaining_len..filled, 0);
                    filled = remaining_len;
                }
                Err(Err::Incomplete(_)) => {
                    // 当前缓冲区没有完整行，需要继续读数据
                    // 如果缓冲区已经满了还没有行结束符，说明行超长，判定为无效
                    if filled == BUF_SIZE {
                        writeln!(invalid_file, "C> {}", String::from_utf8_lossy(&buf[..filled]))?;
                        filled = 0;
                    }
                    break;
                }
                Err(Err::Error(Error { input: _, code })) | Err(Err::Failure(Error { input: _, code })) => {
                    // 解析错误，将当前缓冲区数据判定为无效
                    if filled > 0 {
                        writeln!(invalid_file, "C> {}", String::from_utf8_lossy(&buf[..filled]))?;
                        filled = 0;
                    }
                    if code == ErrorKind::Eof {
                        break;
                    }
                }
            }
        }
    }

    // 处理缓冲区最后剩余的无行结束符的内容
    if filled > 0 {
        let is_valid = buf[..filled].iter().all(|&b| b.is_ascii() && !b.is_ascii_control());
        if is_valid && filled > 0 {
            writeln!(valid_file, "A{}> {}", filled, str::from_utf8(&buf[..filled]).unwrap())?;
        } else {
            writeln!(invalid_file, "C> {}", String::from_utf8_lossy(&buf[..filled]))?;
        }
    }

    Ok(())
}

关键逻辑说明 #

• 缓冲区设计：使用固定1000字节的缓冲区，不会随文件大小增长内存占用，符合大文件流式处理的低内存要求。
• 流式行解析：基于nom 7的streaming::line_ending实现行结束符识别，只有遇到完整行才会触发内容校验和写入，不需要加载全量文件到内存。
• 边界处理：包含了行超长、文件末尾无行结束符、解析异常等场景的处理，不会遗漏任何输入内容。
• 校验逻辑和原有实现完全兼容，仅把打印输出改为写入对应文件，也可以根据需求调整输出格式。

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