Python3 multiprocessing Pool调用对象方法时OpenCV tracker数据未更新

问题根因 #

Python multiprocessing 进程池的核心特性是进程间内存完全隔离，所有传入子进程的参数、调用的对象方法，都会先通过pickle序列化后拷贝到子进程空间，子进程对对象的所有修改都只会作用在本地副本上，永远不会同步回主进程。

你当前的代码中，主进程的重初始化操作、子进程调用t.update()的状态修改，完全作用在两个独立的对象副本上，因此会出现两个核心问题：

• 主进程的重初始化操作不会同步到子进程的tracker副本
• 子进程调用update产生的状态更新也不会回写到主进程的tracker，下次调用进程池时传入的还是主进程里完全未更新的旧tracker对象

即使每次新建进程池也无法解决问题，因为新建进程池每次传入的依然是主进程中从未被更新过的旧tracker副本。

解决方案：使用常驻子进程+消息队列维护tracker状态 #

不要使用单次调用就销毁的进程池，而是为每个tracker启动一个常驻子进程，子进程内部持有自己的tracker对象，主进程通过进程安全的队列和子进程通信，传递重初始化指令、帧数据，子进程执行完操作后把跟踪结果返回给主进程。全程每个tracker的状态只在对应子进程内修改，没有跨进程同步对象状态的需求，自然不会出现状态不一致问题，同时还保留了并行计算的效率优势。

代码实现示例 #

1. 子进程逻辑与初始化

import multiprocessing as mp
import cv2

def tracker_worker(input_queue, output_queue):
    # 子进程内部独立维护tracker对象
    tracker = cv2.TrackerKCF_create()
    is_init = False
    while True:
        cmd, data = input_queue.get()
        if cmd == "init":
            frame, initBB = data
            tracker.clear()
            tracker = cv2.TrackerKCF_create()
            tracker.init(frame, initBB)
            is_init = True
            output_queue.put(("init_done", None))
        elif cmd == "update":
            if not is_init:
                output_queue.put(("update_res", (False, None)))
                continue
            frame = data
            ret, bbox = tracker.update(frame)
            output_queue.put(("update_res", (ret, bbox)))
        elif cmd == "stop":
            break

# 主进程初始化时为每个tracker创建对应子进程和队列
def initTrackers(self, nmbTrackers):
    self.nmbTrackers = nmbTrackers
    self.input_queues = []
    self.output_queues = []
    self.workers = []
    for _ in range(nmbTrackers):
        in_q = mp.Queue()
        out_q = mp.Queue()
        p = mp.Process(target=tracker_worker, args=(in_q, out_q))
        p.start()
        self.input_queues.append(in_q)
        self.output_queues.append(out_q)
        self.workers.append(p)

2. 重初始化方法修改

def reInitTracker(self, index, frame):
    if index >= self.nmbTrackers:
        return

    initBB = cv2.selectROI("Camera view", frame, fromCenter=False,
            showCrosshair=True)
    # 给对应子进程发送初始化指令
    self.input_queues[index].put(("init", (frame, initBB)))
    # 等待初始化完成
    self.output_queues[index].get()

3. 并行更新方法修改

def updateTrackers(self, frame):
    # 给所有子进程批量发送更新指令
    for q in self.input_queues:
        q.put(("update", frame))
    # 批量收集所有跟踪结果
    results = []
    for q in self.output_queues:
        _, res = q.get()
        results.append(res)
    # results中按顺序存储所有tracker的(ret, bbox)结果，可直接使用
    return results

4. 进程退出清理

程序退出前调用该方法销毁所有子进程：

def stopTrackers(self):
    for q in self.input_queues:
        q.put(("stop", None))
    for p in self.workers:
        p.join()

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