树莓派Python多进程：类方法进程间数据共享异常排查

我正在树莓派上开发一个模块，需要通过CNN模型对摄像头捕获的图像进行推理。由于网络耗时及OpenCV的帧缓冲特性，我计划采用Python的multiprocessing模块实现以下逻辑：

• 名为Camera的类维护cv2.VideoCapture()与摄像头的连接，持续采集图像，并将图像存储为可被其他进程访问的形式；
• 主进程以不规则间隔（速度远慢于Camera类的采集速度）获取最新帧。

示例代码如下（省略无关组件）：

import multiprocessing as mp
import cv2
import time  # 补充原代码遗漏的time导入

class Camera():
    def __init__(self):
        self.camera_lock = mp.Lock()
        self.cap = cv2.VideoCapture(0)
        self.last_frame = None  # 存储最后捕获的帧
        self.frame_count = 0  # 捕获帧的计数
        self.camera_process = mp.Process(target=self.camera_read)
        self.camera_process.start()
    
    def camera_read(self):
        while True:
            ret, frame = self.cap.read()
            if ret:
                with self.camera_lock:
                    self.frame_count += 1
                    self.last_frame = frame
                print("Cam process says: frame {} stored".format(self.frame_count))
                time.sleep(0.1)

    def retrieve_last_frame(self):
        with self.camera_lock:
            return self.frame_count  # 测试用，仅返回计数

    def get_frame_count(self):  # 补充原代码中调用的方法
        return self.retrieve_last_frame()

if __name__ == "__main__":
    def frame_consumer(cam, t):
        for i in range(t):
            n = cam.get_frame_count()
            print("Main says: frame {} fetched".format(n))
            time.sleep(2)

    camera = Camera()
    frame_consumer(camera, 20)

运行后发现：子进程能正常采集帧并递增计数器，但主进程每次都输出frame 0，请问问题出在哪里？

核心原因：多进程的内存隔离机制 #

Python的multiprocessing.Process创建子进程时，会复制父进程的所有对象到子进程独立的内存空间。也就是说，子进程中修改的self.frame_count和self.last_frame，只是子进程自己内存里的副本，和主进程中Camera实例的同名变量完全是两个不同的东西。你加的mp.Lock只是同步了子进程内部的操作，但没法让子进程的修改同步到主进程的变量上——因为两者根本不在同一块内存里。

解决方法：使用共享内存对象 #

要实现跨进程的变量共享，必须用multiprocessing提供的共享内存类型，比如Value、Array，或者通过Manager创建可共享的对象。

方案1：用mp.Value共享计数器 #

修改Camera类的初始化和相关方法，把frame_count换成mp.Value（支持原子操作，加锁可进一步避免冲突）：

import multiprocessing as mp
import cv2
import time

class Camera():
    def __init__(self):
        self.camera_lock = mp.Lock()
        self.cap = cv2.VideoCapture(0)
        self.last_frame = None
        # 用mp.Value创建共享整数，'i'表示int类型
        self.frame_count = mp.Value('i', 0)
        self.camera_process = mp.Process(target=self.camera_read)
        self.camera_process.start()
    
    def camera_read(self):
        while True:
            ret, frame = self.cap.read()
            if ret:
                with self.camera_lock:
                    # 操作共享变量需要访问其value属性
                    self.frame_count.value += 1
                    self.last_frame = frame
                print("Cam process says: frame {} stored".format(self.frame_count.value))
                time.sleep(0.1)

    def retrieve_last_frame(self):
        with self.camera_lock:
            return self.frame_count.value, self.last_frame.copy()  # 返回计数和帧的副本

    def get_frame_count(self):
        return self.retrieve_last_frame()[0]

if __name__ == "__main__":
    def frame_consumer(cam, t):
        for i in range(t):
            n = cam.get_frame_count()
            print("Main says: frame {} fetched".format(n))
            time.sleep(2)

    camera = Camera()
    frame_consumer(camera, 20)

方案2：用mp.Manager共享复杂对象（比如图像帧） #

如果要共享last_frame这类numpy数组，mp.Value/Array不够用，可以用mp.Manager创建一个共享的命名空间：

import multiprocessing as mp
import cv2
import time
import numpy as np

class Camera():
    def __init__(self):
        self.camera_lock = mp.Lock()
        self.cap = cv2.VideoCapture(0)
        # 创建Manager，用于共享帧和计数
        self.manager = mp.Manager()
        self.shared_data = self.manager.Namespace()
        self.shared_data.frame_count = 0
        self.shared_data.last_frame = None
        self.camera_process = mp.Process(target=self.camera_read)
        self.camera_process.start()
    
    def camera_read(self):
        while True:
            ret, frame = self.cap.read()
            if ret:
                with self.camera_lock:
                    self.shared_data.frame_count += 1
                    # 存储帧的副本，避免原数据被覆盖
                    self.shared_data.last_frame = frame.copy()
                print("Cam process says: frame {} stored".format(self.shared_data.frame_count))
                time.sleep(0.1)

    def retrieve_last_frame(self):
        with self.camera_lock:
            return self.shared_data.frame_count, self.shared_data.last_frame.copy() if self.shared_data.last_frame is not None else None

    def get_frame_count(self):
        return self.retrieve_last_frame()[0]

if __name__ == "__main__":
    def frame_consumer(cam, t):
        for i in range(t):
            n = cam.get_frame_count()
            print("Main says: frame {} fetched".format(n))
            time.sleep(2)

    camera = Camera()
    frame_consumer(camera, 20)

额外注意点 #

• 共享内存对象的访问要注意同步，mp.Lock依然需要，避免读写冲突；
• 对于图像这类大对象，频繁复制会有性能损耗，可以考虑用multiprocessing.Queue传递最新帧——子进程把最新帧放入队列，主进程每次清空队列取最后一个元素，减少内存占用；
• 树莓派上的摄像头采集要注意帧率和资源占用，避免子进程过度消耗CPU。

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