使用multiprocessing.shared_memory传递datetime对象的Pickle错误排查

错误原因解析 #

• 单个字节位置赋值字节串触发类型错误
  shm.buf是memoryview对象，shm.buf[0]指向共享内存的第一个字节，仅接受单个无符号字节（对应格式'B'），但你把pickle生成的多字节串直接赋值给它，自然会抛出TypeError。
• 主进程逻辑阻塞
  你在主进程直接调用了死循环的create_shared_block()，导致主进程卡在这个函数里，永远执行不到启动子进程的代码。
• 读取逻辑不完整
  就算写入成功，子进程里existing_shm.buf[0]只取了第一个字节，根本无法反序列化出完整的datetime对象。

至于numpy示例能正常运行，是因为numpy内部会批量将数组数据拷贝到共享内存的memoryview中，而非像你这样直接操作单个字节位置，避开了手动赋值的坑。

修正后的代码 #

from datetime import datetime
import time
from multiprocessing import Process, shared_memory
import pickle

def pickle_object(obj):
    return pickle.dumps(obj)

def unpickle_object(bytes_data):
    return pickle.loads(bytes_data)

def write_to_shared_memory(shr_name):
    shm = shared_memory.SharedMemory(name=shr_name, create=True, size=1024)
    try:
        while True:
            pickled_data = pickle_object(datetime.now())
            # 先写入数据长度（用4字节存储，足够覆盖datetime的pickle数据长度）
            shm.buf[:4] = len(pickled_data).to_bytes(4, byteorder='little')
            # 批量写入pickle数据
            shm.buf[4:4+len(pickled_data)] = pickled_data
            print(f"写入的pickle数据: {pickled_data}")
            time.sleep(5)
    finally:
        shm.close()

def read_from_shared_memory(shr_name, process_name):
    shm = shared_memory.SharedMemory(name=shr_name)
    try:
        while True:
            # 先读取数据长度
            length_bytes = shm.buf[:4]
            data_length = int.from_bytes(length_bytes, byteorder='little')
            # 读取对应长度的完整pickle数据
            pickled_data = shm.buf[4:4+data_length]
            dt = unpickle_object(pickled_data)
            print(f"{process_name} 读取到时间: {dt}")
            time.sleep(3 if process_name == "p1" else 2)
    finally:
        shm.close()

if __name__ == "__main__":
    print("启动共享内存写入进程")
    # 将写入逻辑放到子进程，避免阻塞主进程
    writer_process = Process(target=write_to_shared_memory, args=("shared_mem_test",))
    writer_process.start()

    # 启动两个读取进程
    p1 = Process(target=read_from_shared_memory, args=("shared_mem_test", "p1"))
    p2 = Process(target=read_from_shared_memory, args=("shared_mem_test", "p2"))
  
    p1.start()
    p2.start()

    # 等待所有进程结束（手动终止前会持续运行）
    writer_process.join()
    p1.join()
    p2.join()

    # 清理共享内存（手动终止时需额外处理，确保资源释放）
    shm_cleanup = shared_memory.SharedMemory(name="shared_mem_test")
    shm_cleanup.unlink()
    shm_cleanup.close()

关键修正点 #

• 批量写入数据：通过memoryview切片将完整的pickle字节串拷贝到共享内存，同时先写入数据长度，让读取端明确要读取的字节数。
• 写入逻辑独立成子进程：避免主进程被死循环阻塞，确保读取进程能正常启动。
• 完整读取数据：先读取长度再读取对应字节，保证拿到完整的序列化数据进行反序列化。

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