为了实现基于 `QSharedMemory` 的多进程之间的数据读写功能，并将其封装成类以便复用，我们可以设计一个通用的模板。以下是详细的步骤及其实现。 --- ### 封装思路 1. **核心组件** - 使用 PyQt 或 PySide 提供的 `QSharedMemory` 来管理共享内存。 - 利用 NumPy 数组或其他序列化工具（如 pickle）将复杂的数据结构转换为字节数组，方便存入共享内存。 2. **关键点** - 创建或附加到现有的共享内存段。 - 数据写入时确保正确填充内容长度字段。 - 读取数据前校验完整性以防止竞争条件引发的问题。 3. **同步机制**（可选） - 因为 `QSharedMemory` 自身并不提供锁保护措施，所以可能还需要借助信号量(`QLock`)等互斥手段保障一致性和安全性。 下面是具体的 Python 实现阶段: --- ### 完整代码示例 ```python from PyQt5.QtCore import QSharedMemory, QByteArray import numpy as np import sys class SharedMemoryManager: def __init__(self, key: str, size: int = 1024): """ 初始化共享内存管理者 :param key: 共享内存唯一标识符(字符串形式) :param size: 分配的空间大小，默认值足够存放一些小规模数组 """ self.key = key self.size = max(size + 64, 128) # 额外增加头部空间用于元信息存储 self.shared_memory = QSharedMemory(key) def attach_or_create(self)->bool: """尝试连接已存在的共享内存或者新建""" if not self.shared_memory.attach(QSharedMemory.ReadOnly|QSharedMemory.ReadWrite): success=self.shared_memory.create(self.size) return bool(success) else : return True def write_data(self,data_array,np_type='float'): """把numpy array 写进指定位置""" dtype=np.dtype(np_type).newbyteorder('>') serialized_bytes=data_array.astype(dtype).tobytes() actual_len=len(serialized_bytes) total_size=min(actual_len+8,self.size)#限制不超过预设容量上限 buffer=QByteArray(b'\x00'*total_size) view=buffer.data() struct.pack_into('>Q',view,0,total_size-8 ) view[8:(actual_len+8)] =serialized_bytes[:] if not self.shared_memory.lock(): raise Exception("无法加锁") try: self.shared_memory.write(buffer) finally: self.shared_memory.unlock() def read_data(self,dtype:str='float')->np.ndarray: """从共享区提取有效负载返回给用户端""" if not self.shared_memory.isAttached(): attached_flag=self.shared_memory.attach(QSharedMemory.ReadOnly | QSharedMemory.ReadWrite) if not attached_flag:raise RuntimeError('Attach Failed') raw_content=QByteArray() locked=False try: if not self.shared_memory.lock(): raise ChildProcessError('Lock timeout.') else: locked=True raw_content=self.shared_memory.read(self.shared_memory.size()) expected_payload_length,*rest=struct.unpack_from('>Q',raw_content.bytes(),offset=0) deserialized_buffer=memoryview(raw_content)[8:] restored_ndarray=np.frombuffer(deserialized_buffer,dtype=dtype,newshape=(-1,)) return restored_ndarray[:-expected_payload_length%deserialized_buffer.itemsize].copy(order='C') except IndexError as e: print(f"Read failed due malformed structure:{e}") finally: if locked:self.shared_memory.unlock() def cleanup_resources(self): del self.shared_memory if __name__ == '__main__': sm_manager_a = SharedMemoryManager('my_unique_key_123',size=4*1024**2) test_arr=np.random.rand(1000,) status_created=sm_manager_a.attach_or_create() if status_created: sm_manager_a.write_data(test_arr,'double') another_process_instance_b=SharedMemoryManager('my_unique_key_123',) retrieved_values=another_process_instance_b.read_data('double') assert all(retrieved_values==test_arr), "Data integrity check failure." both_done=[sm_manager.cleanup_resources()for sm_manager in [sm_manager_a ,another_process_instance_b]] ``` 上述例子展示了如何利用Qt提供的低级接口完成高级别的任务流程控制以及异常情况下的优雅降级策略.