如何在Cython中为跨进程对象授予只读（非复制）访问权限？

针对你提出的两个问题——本质都是如何在Cython中实现跨进程的只读、非复制对象访问，核心思路是绕开Python进程的内存隔离，用共享内存结合Cython的底层内存操作来实现。下面结合你给出的Process1/2/3的场景，一步步拆解方案：

核心原理：利用共享内存打破进程隔离 #

Python（包括Cython）的进程默认是完全隔离的，每个进程有自己的内存空间，直接共享Python对象不可能。但我们可以把需要共享的结构化数据（比如Tasks类的核心字段）放到一块操作系统级的共享内存里，让所有进程直接访问这块物理内存，这样就避免了复制。同时用Cython封装访问逻辑，严格限制只读权限。

第一步：在Process1中初始化共享内存并写入数据 #

首先，我们需要把Tasks类的核心数据转换成C级别的结构体（这样Cython可以直接解析，没有序列化开销），然后写入共享内存。

比如先在Cython里定义任务的结构体：

cdef struct TaskData:
    int task_id               # 任务ID
    char task_name[64]        # 任务名称（固定长度，避免内存碎片）
    bint is_completed         # 任务完成状态

然后在Process1的Tasks类里，添加初始化共享内存的方法：

from multiprocessing.shared_memory cimport SharedMemory
from libc.string cimport memcpy

cdef class Tasks:
    cdef list _tasks  # Process1内部的任务列表
    cdef SharedMemory _shm
    cdef TaskData* _shared_tasks
    cdef int _task_count

    def __init__(self):
        self._tasks = []
        # 后续可以添加任务的方法，比如add_task等

    def init_shared_memory(self, int task_count):
        self._task_count = task_count
        # 创建共享内存，大小是单个TaskData的尺寸乘以任务数
        self._shm = SharedMemory(create=True, size=sizeof(TaskData)*task_count)
        # 把共享内存映射成C指针，方便直接操作
        self._shared_tasks = <TaskData*>self._shm.buf
        # 把Process1内部的任务数据写入共享内存
        self._sync_to_shared()

    cdef _sync_to_shared(self):
        # 同步Process1的任务数据到共享内存（仅Process1可调用）
        for i in range(self._task_count):
            task = self._tasks[i]
            self._shared_tasks[i].task_id = task["id"]
            # 把字符串拷贝到C结构体的char数组里
            memcpy(self._shared_tasks[i].task_name, task["name"].encode('utf-8'), len(task["name"])+1)
            self._shared_tasks[i].is_completed = task["is_completed"]

    # 提供给Process1的写方法，写完后同步到共享内存
    def update_task(self, int task_id, bint completed):
        # 先修改Process1内部的_tasks
        for task in self._tasks:
            if task["id"] == task_id:
                task["is_completed"] = completed
                break
        # 同步到共享内存
        self._sync_to_shared()

    # 获取共享内存的名称，传递给其他进程
    property shared_memory_name:
        def __get__(self):
            return self._shm.name

第二步：在Process2/3中封装只读访问类 #

在子进程里，我们只需要挂载已有的共享内存，然后提供纯只读的接口，绝对禁止任何写操作：

from multiprocessing.shared_memory cimport SharedMemory

cdef class ReadOnlyTasks:
    cdef SharedMemory _shm
    cdef TaskData* _shared_tasks
    cdef int _task_count

    def __init__(self, str shm_name, int task_count):
        # 挂载Process1创建的共享内存（不创建新的）
        self._shm = SharedMemory(name=shm_name, create=False)
        self._shared_tasks = <TaskData*>self._shm.buf
        self._task_count = task_count

    # 只读方法：获取单个任务的信息
    def get_task(self, int idx):
        if idx < 0 or idx >= self._task_count:
            raise IndexError("Task index out of range")
        cdef TaskData* task_ptr = &self._shared_tasks[idx]
        return {
            "task_id": task_ptr.task_id,
            "task_name": task_ptr.task_name.decode('utf-8'),
            "is_completed": bool(task_ptr.is_completed)
        }

    # 禁止任何修改操作：重载__setattr__拦截非法赋值
    def __setattr__(self, name, value):
        # 只允许初始化时设置内部的共享内存相关属性
        if name not in ("_shm", "_shared_tasks", "_task_count"):
            raise PermissionError("ReadOnlyTasks does not allow any modification")
        super().__setattr__(name, value)

    # 可选：实现迭代器，方便遍历所有任务
    def __iter__(self):
        for i in range(self._task_count):
            yield self.get_task(i)

第三步：跨进程传递共享内存信息 #

Process1初始化共享内存后，把它的名称（tasks.shared_memory_name）和任务数量传递给Process2和Process3。可以用multiprocessing.Queue或者启动子进程时作为参数传入：

比如用multiprocessing.Process启动子进程：

from multiprocessing import Process
import tasks_cython  # 编译后的Cython模块

def run_process2(shm_name, task_count):
    ro_tasks = tasks_cython.ReadOnlyTasks(shm_name, task_count)
    # 示例：遍历所有任务
    for task in ro_tasks:
        print(f"Process2 read task: {task}")

if __name__ == "__main__":
    # Process1初始化Tasks
    main_tasks = tasks_cython.Tasks()
    # 添加一些任务
    main_tasks._tasks = [
        {"id": 1, "name": "Job A", "is_completed": False},
        {"id": 2, "name": "Job B", "is_completed": True}
    ]
    main_tasks.init_shared_memory(task_count=2)

    # 启动Process2和Process3
    p2 = Process(target=run_process2, args=(main_tasks.shared_memory_name, 2))
    p3 = Process(target=run_process2, args=(main_tasks.shared_memory_name, 2))
    p2.start()
    p3.start()
    p2.join()
    p3.join()

    # 最后销毁共享内存
    main_tasks._shm.close()
    main_tasks._shm.unlink()

关键注意事项 #

1. 同步与一致性：如果Process1会动态更新数据，必须加同步锁！比如用multiprocessing.Lock，Process1写之前加锁，Process2/3读之前也加锁（或者用读写锁RLock，允许多个读操作同时进行，提高效率）。
2. 内存生命周期：共享内存需要由创建它的Process1最后销毁，否则会残留内存垃圾。务必调用shm.close()和shm.unlink()。
3. Cython编译：要确保你的setup.py正确配置，比如：

from setuptools import setup
from Cython.Build import cythonize

setup(
    name="tasks_cython",
    ext_modules=cythonize("tasks.pyx"),
)

然后用python setup.py build_ext --inplace编译成扩展模块。

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