如何用Rust Serde在后台线程反序列化大内存占用数组

核心思路 #

由于Deserializer和SeqAccess未实现Send，无法跨线程传递反序列化上下文，所以正确的做法是：

• 在发起反序列化的线程中逐个流式读取数组元素，完成反序列化后立刻通过通道发送给后台线程
• 后台线程仅负责接收已反序列化完成的元素并处理，不触碰反序列化核心上下文

具体实现步骤 #

1. 定义结构体与自定义数组类型 #

把原本存储Vec<T>的字段替换为自定义逻辑的接收器类型，告知serde使用自定义反序列化：

use serde::{Deserialize, Deserializer};
use crossbeam_channel::{unbounded, Receiver, Sender};
use std::thread;

// 包含超大数组的外部结构体
#[derive(Deserialize)]
struct OuterStruct {
    a: String,
    #[serde(deserialize_with = "deserialize_streamed_array")]
    c: Receiver<InnerItem>,
}

// 数组内部元素结构
#[derive(Deserialize, Debug)]
struct InnerItem {
    id: u64,
    content: String,
}

2. 实现自定义反序列化函数 #

在主线程中遍历数组元素，反序列化后发送到通道，同时返回接收器给外部调用方：

fn deserialize_streamed_array<'de, D, T>(deserializer: D) -> Result<Receiver<T>, D::Error>
where
    D: Deserializer<'de>,
    T: Deserialize<'de> + Send + 'static,
{
    // 创建支持多消费者的无界通道
    let (sender, receiver) = unbounded();

    struct StreamedArrayVisitor<T> {
        sender: Sender<T>,
    }

    impl<'de, T> serde::de::Visitor<'de> for StreamedArrayVisitor<T>
    where
        T: Deserialize<'de>,
    {
        type Value = ();

        fn expecting(&self, formatter: &mut std::fmt::Formatter) -> std::fmt::Result {
            formatter.write_str("JSON数组")
        }

        fn visit_seq<S>(self, mut seq: S) -> Result<Self::Value, S::Error>
        where
            S: serde::de::SeqAccess<'de>,
        {
            // 逐个读取并反序列化数组元素，发送到通道
            while let Some(item) = seq.next_element()? {
                // 若接收器已销毁（消费线程退出），停止处理
                if self.sender.send(item).is_err() {
                    break;
                }
            }
            Ok(())
        }
    }

    // 启动两个后台消费线程
    spawn_consumers(receiver.clone());

    // 执行流式反序列化
    deserializer.deserialize_seq(StreamedArrayVisitor { sender })?;

    Ok(receiver)
}

3. 多线程消费逻辑 #

利用crossbeam-channel支持多消费者的特性，克隆接收器分配给不同线程：

fn spawn_consumers<T>(receiver: Receiver<T>)
where
    T: Send + 'static + std::fmt::Debug,
{
    // 消费者线程1
    thread::spawn(move || {
        for item in receiver.clone() {
            println!("消费者1处理元素: {:?}", item);
            // 替换为你的业务逻辑
        }
    });

    // 消费者线程2
    thread::spawn(move || {
        for item in receiver {
            println!("消费者2处理元素: {:?}", item);
            // 替换为你的业务逻辑
        }
    });
}

4. 最终调用示例 #

使用流式读取避免加载整个文件到内存，反序列化同时后台线程自动处理数组元素：

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let file = std::fs::File::open("large_data.json")?;
    let reader = std::io::BufReader::new(file);

    // 流式反序列化外部结构体，此时后台线程已开始处理数组元素
    let outer: OuterStruct = serde_json::from_reader(reader)?;

    println!("外部结构体字段a: {}", outer.a);
    // 主线程可继续执行其他任务，无需等待数组处理完成

    // 示例：等待所有元素处理完毕
    std::thread::sleep(std::time::Duration::from_secs(15));
    Ok(())
}

关键注意事项 #

• 禁止跨线程传递Deserializer/SeqAccess：这两个 trait 未强制要求实现Send，大部分serde反序列化器（如serde_json）都不支持跨线程，必须在发起反序列化的线程内完成元素读取与反序列化。
• 通道选型：原生std::sync::mpsc不支持多消费者，crossbeam-channel是更适合多线程消费场景的选择。
• 优雅终止：若消费线程提前退出，发送方会收到错误，此时可停止处理剩余数组元素，避免无效计算。

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