树莓派Python2.7多进程间共享值更新不生效问题求助

嘿，看起来你碰到了Python多进程共享状态不同步的典型问题，我来帮你拆解一下问题根源，然后一步步解决：

1. 共享状态类的核心错误：类属性误用 #

你的SharedImageDetectionValue类把stored_has_position_found定义成了类属性，而不是实例属性——这是导致状态无法同步的关键！

类属性属于整个类，而multiprocessing.Manager的代理机制是针对实例属性做跨进程同步的。当你通过Manager创建共享实例时，其他进程访问的是这个实例的代理对象，类属性的修改无法被正确同步到代理中；而且当你调用set_has_position_found时，会给实例添加一个局部的实例属性，覆盖类属性，但这个修改不会传递到其他进程。

修复方法： #

把共享状态改成实例属性，在__init__里初始化：

class SharedImageDetectionValue(object): 
    def __init__(self):
        # 改为实例属性，确保Manager能正确同步跨进程的状态变更
        self.stored_has_position_found = False 

    def get_has_position_found(self): 
        return self.stored_has_position_found 

    def set_has_position_found(self, new_value): 
        self.stored_has_position_found = new_value 
        return self.stored_has_position_found 

2. 步进电机进程的运行问题 #

你提到把Stepper放进进程池后无法正常工作，这其实是树莓派硬件控制的常见限制：步进电机的GPIO操作通常需要在主进程（或拥有独占硬件访问权限的进程）中运行。进程池的子进程可能无法正确访问GPIO资源，或者会和其他进程产生资源竞争，导致电机控制失效。

建议： #

保持Stepper在主进程中运行（就像你现在的代码结构），只要修复了共享状态的问题，就能正常读取到图像处理进程的状态变更。如果一定要用子进程运行步进电机，需要确保：

• 子进程在创建前已经完成GPIO初始化
• 避免多个进程同时操作GPIO（步进电机控制是独占式的）

3. 验证异步进程是否执行 #

你的pool.apply_async是异步执行的，有时候可能会因为进程池的调度问题，导致图像处理进程没有及时启动。可以添加一个回调函数来确认它是否真的执行了：

def on_process_done(result):
    print("[ MAIN ] 图像处理完成，已标记目标找到")

pool.apply_async(imgProcessor.run, args=(shared, ), callback=on_process_done)

这样能直观看到图像处理进程是否完成了状态设置。

完整修复后的示例代码 #

我把修复后的完整代码整理好了，你可以直接测试：

from multiprocessing.managers import BaseManager
from multiprocessing import Pool

class SharedImageDetectionValue(object): 
    def __init__(self):
        self.stored_has_position_found = False 

    def get_has_position_found(self): 
        return self.stored_has_position_found 

    def set_has_position_found(self, new_value): 
        self.stored_has_position_found = new_value 
        return self.stored_has_position_found 

class ImageProcessor: 
    def __init__(self): 
        pass

    def run(self, shared): 
        # 测试用：直接设置为找到目标
        print("[ 图像处理进程 ] 标记目标已找到")
        shared.set_has_position_found(True)
        return True

class Stepper: 
    def __init__(self): 
        self.running = True

    def do_steps(self, delay):
        # 模拟步进电机动作
        print("[ 步进电机进程 ] 执行步进...")
        import time
        time.sleep(delay)

    def run_until_stopped(self, shared): 
        print("[ 步进电机进程 ] 启动电机循环")
        while self.running: 
            has_found = shared.get_has_position_found() 
            print("[ 步进电机进程 ] 目标是否找到: " + str(has_found)) 
            if not has_found: 
                self.do_steps(0.0005)
            else:
                print("[ 步进电机进程 ] 检测到目标，停止电机")
                self.running = False

class ProcessManager(BaseManager): 
    pass

if __name__ == "__main__":
    # 必须在主模块中执行多进程逻辑，避免子进程重复初始化
    ProcessManager.register('SharedImageDetectionValue', SharedImageDetectionValue) 
    manager = ProcessManager() 
    manager.start() 
    shared = manager.SharedImageDetectionValue() 

    pool = Pool(1) 
    imgProcessor = ImageProcessor() 
    # 添加回调确认图像处理进程执行状态
    pool.apply_async(imgProcessor.run, args=(shared, ), callback=on_process_done) 

    print("[ 主进程 ] 启动步进电机循环")
    Stepper().run_until_stopped(shared) 

    # 清理资源
    pool.close()
    pool.join()
    manager.shutdown()

额外注意事项 #

• 一定要把主逻辑放在if __name__ == "__main__":块里，这是Python多进程的强制规范，避免子进程重复初始化导致的奇怪问题。
• 树莓派的GPIO操作需要权限，确保运行脚本的用户属于gpio组，或者用sudo运行。
• 如果你的Stepper继承了Observable，要确保这个类的多线程/多进程兼容性，别引入额外的同步问题。

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