Python多进程中自定义数据类型的微秒级数据检索方案咨询

针对多进程下含自定义类型（如QPointF）列表的微秒级数据检索需求，以下是两种高效可行的方案：

1. 基于ctypes + multiprocessing.Array的共享内存方案 #

QPointF本质是由两个64位浮点数（x、y）组成的结构体，我们可以用ctypes定义对应的数据结构，再通过multiprocessing.Array创建共享内存数组，实现进程间直接访问，无需额外序列化/反序列化操作。

实现步骤： #

• 定义与QPointF匹配的ctypes结构体：

import ctypes
from multiprocessing import Process, Array

# 对应QPointF的x、y双精度浮点字段
class CPointF(ctypes.Structure):
    _fields_ = [("x", ctypes.c_double), ("y", ctypes.c_double)]

• 主进程创建共享数组并写入数据：

def main_process():
    # 创建可容纳10000个CPointF的共享数组
    shared_array = Array(CPointF, 10000)
    # 写入数据（直接操作共享内存）
    for i in range(10000):
        shared_array[i].x = i
        shared_array[i].y = 100.0
    # 启动子进程并传递共享数组
    p = Process(target=sub_process, args=(shared_array,))
    p.start()
    p.join()

• 子进程直接检索使用数据：

def sub_process(shared_array):
    # 微秒级检索，直接访问共享内存中的数据
    point = shared_array[5000]
    # 直接使用，无需额外转换
    print(f"Retrieved point: ({point.x}, {point.y})")

该方案核心是利用共享内存直接存储QPointF的二进制结构，进程间访问无需拷贝或序列化，检索速度可达微秒级，且数据可直接使用。

2. 基于mmap的内存映射文件方案 #

如果需要更大规模的数据共享，可使用mmap模块创建内存映射文件，将QPointF数据按二进制格式写入，子进程直接映射后解析为QPointF对象，同样无需额外处理。

实现示例： #

import mmap
import os
from PyQt5.QtCore import QPointF

def write_mmap_data():
    # 创建内存映射文件，大小为10000个QPointF（每个占16字节：2*8）
    file_size = 10000 * 16
    with open("shared_data.bin", "wb") as f:
        f.write(b'\x00' * file_size)
    
    with open("shared_data.bin", "r+b") as f:
        with mmap.mmap(f.fileno(), length=file_size, access=mmap.ACCESS_WRITE) as mm:
            # 写入数据
            for i in range(10000):
                qp = QPointF(i, 100.0)
                # 将QPointF的二进制数据写入内存映射
                mm.seek(i * 16)
                mm.write(qp.x().to_bytes(8, byteorder='little'))
                mm.write(qp.y().to_bytes(8, byteorder='little'))

def read_mmap_data():
    file_size = 10000 * 16
    with open("shared_data.bin", "r+b") as f:
        with mmap.mmap(f.fileno(), length=file_size, access=mmap.ACCESS_READ) as mm:
            # 微秒级检索第5000个点
            mm.seek(5000 * 16)
            x = float.from_bytes(mm.read(8), byteorder='little')
            y = float.from_bytes(mm.read(8), byteorder='little')
            qp = QPointF(x, y)
            print(f"Retrieved point: ({qp.x()}, {qp.y()})")

注意事项： #

• 无论使用哪种方案，都需要通过multiprocessing.Lock等同步机制保护共享内存的读写操作，避免数据竞争。
• 若使用PyQt的QPointF，确保多进程中PyQt的初始化正确（子进程需单独初始化QApplication）。

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