Windows下Python3独立进程共享内存：ctypes.Structure数组读写需求

哈哈，这个需求我刚好踩过坑！在Windows下用mmap结合ctypes.Structure确实是实现复杂类型跨进程共享内存的最优解，而且完全支持单个元素的读写更新，不用把整个数组都序列化一遍。我给你一步步拆解实现过程，附带可运行的代码示例：

mmap负责开辟跨进程可见的共享内存块，ctypes.Structure则把这块内存直接映射成我们自定义的结构化数据——相当于给共享内存套了个“Python对象壳”，我们可以像操作普通数组元素一样读写，底层的修改会自动同步到共享内存，其他进程能实时看到变化。

首先得定义一个能在共享内存里正常解析的结构体，重点是用C兼容的基础类型，尤其是字符串不能用Python原生字符串，得转成固定长度的字节数组（避免指针无效的问题）：

import ctypes
import mmap
import os
from multiprocessing import Process

# 定义共享内存中使用的结构体
class SharedData(ctypes.Structure):
    _fields_ = [
        ("id", ctypes.c_int),          # 整数类型
        ("value", ctypes.c_float),     # 浮点类型
        ("name", ctypes.c_char * 32)   # 固定长度字符串（32字节，含结束符）
    ]

划重点：别用ctypes.c_char_p存字符串！它本质是个内存指针，只在当前进程有效，其他进程读了就是个无效地址。固定长度的c_char * N是直接把字节存在共享内存里，所有进程都能正常解析。

接下来创建共享内存块，然后把它映射成我们定义的结构体数组：

def create_shared_memory(array_size):
    # 计算共享内存总大小：单个结构体大小 × 数组元素个数
    buffer_size = ctypes.sizeof(SharedData) * array_size
    # Windows下用tagname创建匿名共享内存，其他进程通过这个名字就能找到它
    shm = mmap.mmap(-1, buffer_size, tagname="MySharedArray")
    # 将共享内存直接映射为结构体数组
    shared_array = (SharedData * array_size).from_buffer(shm)
    return shm, shared_array

直接操作数组索引对应的结构体属性就行，修改会自动同步到共享内存：

def update_element(shared_array, index, data):
    # 直接赋值结构体字段，无需手动写回内存
    shared_array[index].id = data["id"]
    shared_array[index].value = data["value"]
    # 字符串转字节，截断到定义的长度并补结束符
    shared_array[index].name = data["name"].encode("utf-8")[:31] + b"\x00"

同样直接访问索引，把C类型转成Python常用类型即可：

def read_element(shared_array, index):
    elem = shared_array[index]
    return {
        "id": elem.id,
        "value": elem.value,
        "name": elem.name.decode("utf-8").strip("\x00")  # 去掉末尾的结束符
    }

写个主进程创建共享内存、子进程更新元素、主进程读取结果的完整例子：

def child_process_task(index):
    # 子进程通过tagname打开已有的共享内存
    shm = mmap.mmap(-1, 0, tagname="MySharedArray")
    array_size = 10
    # 映射为结构体数组
    shared_array = (SharedData * array_size).from_buffer(shm)
    # 更新指定索引的元素
    update_element(shared_array, index, {"id": index+1, "value": (index+1)*0.5, "name": f"Item_{index+1}"})
    shm.close()

if __name__ == "__main__":
    array_size = 10
    # 主进程创建共享内存和结构体数组
    shm, shared_array = create_shared_memory(array_size)
    
    # 初始化前3个元素
    for i in range(3):
        update_element(shared_array, i, {"id": i, "value": i*0.1, "name": f"Init_Item_{i}"})
    
    # 创建子进程更新第5个元素（索引4）
    p = Process(target=child_process_task, args=(4,))
    p.start()
    p.join()
    
    # 读取更新后的元素
    print("读取索引4的元素：", read_element(shared_array, 4))
    # 读取初始化的元素
    print("读取索引1的元素：", read_element(shared_array, 1))
    
    shm.close()

• 结构体定义一致性: 所有进程必须使用完全相同的SharedData定义，字段顺序、类型、长度不能有差异，否则会出现内存解析错误。
• 并发安全: 如果多个进程同时写同一个元素，需要加锁（比如用multiprocessing.Lock或者Windows原生互斥量），避免数据竞争。
• 共享内存生命周期: 所有进程关闭mmap对象后，共享内存会被系统自动回收；如果需要持久化，可以用文件映射的mmap（指定文件句柄而非-1）。

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