如何使用marshal序列化的现有字节缓冲区创建SharedMemory实例？

下面是几个可落地的方案，你可以根据自己的场景选择：

方案1：先用空写入统计精确序列化大小，再创建对应大小的共享内存写入 #

这个方案完全避免预估大小不准的问题，且第一次统计大小的过程几乎无额外内存开销：

• 先自定义一个仅统计写入字节数的工具类，实现write方法适配marshal的写入接口
• 第一次调用marshal.dump写入该工具类，得到序列化后的精确大小
• 按精确大小创建SharedMemory实例，第二次调用marshal.dump直接写入共享内存缓冲区

代码示例：

import marshal
from multiprocessing.shared_memory import SharedMemory

class SizeCounter:
    def __init__(self):
        self.total = 0
    def write(self, data: bytes):
        self.total += len(data)

# 你的数据对象
data = {'foo': 1, 'bar': 'some text'}

# 第一步：统计序列化所需的精确大小，无额外内存占用
counter = SizeCounter()
marshal.dump(data, counter)

# 第二步：创建刚好大小的共享内存，直接写入序列化结果
shm = SharedMemory(create=True, size=counter.total)
marshal.dump(data, shm.buf.obj)

这个方案唯一的开销是序列化两次相同的对象，对于绝大多数场景，序列化的CPU开销远低于1GB级内存冗余的开销，可接受度很高。注意使用该方案时两次序列化之间不要修改data对象，避免两次结果不一致。

方案2：优化现有复制逻辑，降低额外内存的留存时间 #

如果你无法接受两次序列化的CPU开销，可以优化现有代码的内存使用效率：

• 用切片赋值代替逐字节循环，大幅提升复制速度
• 完成复制后立即手动删除data_bytes变量并触发垃圾回收，减少多余内存的留存时间

代码示例：

import gc
import marshal
from multiprocessing.shared_memory import SharedMemory

data = {'foo': 1, 'bar': 'some text'}
data_bytes = marshal.dumps(data)
shm = SharedMemory(create=True, size=len(data_bytes))
# C层面直接批量复制，比逐字节循环快上百倍
shm.buf[:] = data_bytes
# 立即释放序列化后的字节对象内存
del data_bytes
gc.collect()

这个方案在复制完成后会马上释放多余的1GB内存，峰值内存仍然会达到2倍，但持续时间极短，对于内存余量足够的场景完全可用。

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