Python Pickle 流式序列化实现：避免内存数据重复拷贝方案

你现有方案的核心问题是pickle.dumps会生成完整的序列化字节串，再拷贝到BytesIO中，导致同一份数据在内存中存了两份，大对象场景下内存占用极高。下面是无额外内存冗余的流式实现方案：

方案：基于管道的流式桥接 #

利用操作系统管道做读写两端的缓冲，后台线程执行序列化写操作，前端直接返回可读指针给上传接口，全程无全量数据内存拷贝：

import os
import pickle
import threading
from io import PipeReader, PipeWriter

def get_streaming_pickle_fp(obj):
    # 创建操作系统管道，分为读、写两个端
    read_fd, write_fd = os.pipe()
    read_fp = PipeReader(read_fd)
    write_fp = PipeWriter(write_fd)

    def _dump_worker():
        try:
            # 直接将序列化结果写入管道写端
            pickle.dump(obj, write_fp)
        finally:
            # 序列化完成后必须关闭写端，否则读端会永久阻塞
            write_fp.close()
    
    # 启动后台序列化线程，不阻塞主流程
    threading.Thread(target=_dump_worker, daemon=True).start()
    # 直接返回读端给上传接口使用
    return read_fp

# 业务侧调用方式
fp = get_streaming_pickle_fp(your_obj)
upload_from_file(fp)
# 上传完成后关闭读端释放资源
fp.close()

方案说明 #

• 内存占用仅为操作系统管道默认缓冲区大小（通常为64KB），无论序列化对象多大，都不会出现全量数据冗余存储的问题
• 管道的读写是内核级实现，性能比用户态的内存拷贝高很多
• 后台线程设为daemon=True，即使上传流程意外中断，序列化线程也会自动退出，不会产生僵尸线程

无管道简化方案（适合小对象场景） #

如果你的上传接口支持传入字节迭代器作为输入，可以用更简单的实现，不需要依赖系统管道：

import pickle
import threading
from collections import deque

class PickleIterator:
    def __init__(self, obj):
        self._buf = deque()
        self._worker = threading.Thread(target=lambda: pickle.dump(obj, self), daemon=True)
        self._worker.start()
    
    def write(self, b):
        self._buf.append(b)
    
    def __iter__(self):
        while self._worker.is_alive() or self._buf:
            if self._buf:
                yield self._buf.popleft()

# 调用方式（如果upload_from_file支持迭代器输入）
upload_from_file(PickleIterator(your_obj))

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