Python中如何用ProcessPoolExecutor让子进程持续更新共享变量

核心问题点 #

1. 多进程全局变量不共享：每个子进程有独立内存空间，global关键字无法跨进程同步变量，必须用进程间共享数据结构。
2. 函数调用顺序错误：原代码先提交任务后定义函数，Python会直接报错找不到函数。
3. 循环逻辑混乱：CheckPositionImage里嵌套的while循环逻辑错误，导致要么提前终止要么死循环。
4. 未定义变量：startingtime未初始化，主进程打印会报错。
5. 冗余模块导入：用importlib导入cv2和concurrent完全没必要，直接导入即可。

修正后的代码 #

import pyautogui
import time
import cv2
import numpy
from multiprocessing import Manager, Process

def CheckPositionInput(shared_data):
    while not shared_data['AreaOnScreen']:
        print('Searching for Area.png...')
        pos = pyautogui.locateCenterOnScreen('Area.png', confidence=0.8)
        if pos is not None:
            shared_data['AreaOnScreen'] = True
            shared_data['XMark'], shared_data['YMark'] = pos
            print(f"Found Area at: ({shared_data['XMark']}, {shared_data['YMark']})")
        time.sleep(0.5)  # 降低检测频率，减少资源占用

def CheckPositionImage(shared_data):
    print('Waiting for Area to be found...')
    # 等待Area被检测到
    while not shared_data['AreaOnScreen']:
        time.sleep(0.2)
    
    print('Start tracking Image.png...')
    # 持续检测直到图像消失
    while True:
        pos = pyautogui.locateCenterOnScreen('Image.png', confidence=0.6)
        if pos is not None:
            shared_data['XMark'], shared_data['YMark'] = pos
            print(f"Current Image Y: {shared_data['YMark']}")
        else:
            print('Image.png disappeared, stopping tracking.')
            shared_data['ImageAlive'] = False
            break
        time.sleep(0.3)  # 控制检测间隔

if __name__ == '__main__':
    startingtime = time.time()
    # 使用Manager创建跨进程共享的字典
    with Manager() as manager:
        shared_data = manager.dict()
        # 初始化共享变量
        shared_data['AreaOnScreen'] = False
        shared_data['XMark'] = 0
        shared_data['YMark'] = 0
        shared_data['ImageAlive'] = True

        # 创建子进程
        p1 = Process(target=CheckPositionInput, args=(shared_data,))
        p2 = Process(target=CheckPositionImage, args=(shared_data,))

        # 启动进程
        p1.start()
        p2.start()

        # 等待图像跟踪进程结束
        p2.join()
        # 终止Area检测进程（因为已经不需要了）
        p1.terminate()

        # 打印最终结果
        print(f"Final YMark: {shared_data['YMark']}")
        print(f"Total runtime: {time.time() - startingtime:.2f} seconds")

关键修改说明 #

• 进程间共享数据：用multiprocessing.Manager创建共享字典，所有进程操作这个字典里的变量，实现跨进程同步。
• 修复循环逻辑：CheckPositionImage先等待AreaOnScreen变为True，然后进入循环检测图像，直到图像消失就退出循环。
• 调整函数顺序：先定义函数再创建进程，避免Python报错。
• 添加延迟：在检测循环里加入time.sleep()，减少CPU占用，同时避免检测过于频繁导致的问题。
• 正确的进程管理：用Process类创建进程，主进程等待跟踪进程结束后，终止Area检测进程，避免资源浪费。

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