如何在SharedMemory与BytesIO之间复制字节数据？

1. 从 SharedMemory 复制到 BytesIO #

利用SharedMemory的buf内存视图，直接将其转换为bytes后写入BytesIO即可，操作高效直观：

from multiprocessing import shared_memory
import io

# 连接到已存在的共享内存对象（假设名称为"my_shared_mem"）
shm = shared_memory.SharedMemory(name="my_shared_mem", create=False)
# 将内存视图转换为bytes并初始化BytesIO
bytes_io = io.BytesIO(bytes(shm.buf))

# 后续可像操作普通文件一样使用bytes_io
bytes_io.seek(0)
processed_data = bytes_io.read()

# 使用完毕关闭共享内存连接
shm.close()

如果只需复制共享内存中的部分数据，可通过内存视图切片实现：

# 仅复制前1024字节
partial_data = bytes(shm.buf[:1024])
bytes_io = io.BytesIO(partial_data)

2. 从 BytesIO 复制到 SharedMemory #

需确保共享内存的容量足够容纳BytesIO中的数据，然后将BytesIO的内容写入内存视图：

from multiprocessing import shared_memory
import io

# 初始化并填充BytesIO对象
bytes_io = io.BytesIO(b"待写入共享内存的原始数据")
bytes_io.seek(0)
data_to_write = bytes_io.read()

# 创建或连接到指定大小的共享内存对象
shm = shared_memory.SharedMemory(name="my_shared_mem", create=True, size=len(data_to_write))
# 将数据写入共享内存的内存视图
shm.buf[:len(data_to_write)] = data_to_write

# 使用完毕关闭并释放共享内存（仅创建者需调用unlink）
# shm.close()
# shm.unlink()

注意事项 #

• 共享内存的大小必须大于等于待写入的数据长度，否则会抛出ValueError。
• 多进程并发操作共享内存时，需配合锁（如multiprocessing.Lock）避免读写冲突。
• 内存视图的切片赋值操作直接操作底层内存，无需额外数据拷贝，性能优于传统文件读写。

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