使用多进程加速OpenCV视频处理时遇pickle线程锁对象报错

问题根源 #

你遇到的TypeError: can't pickle _thread.lock objects，本质是因为multiprocessing进程间通信需要对参数做序列化（pickle），但你传入的car_tracker和ped_tracker（Sort算法实例）内部包含了线程锁对象——这类对象是无法被pickle序列化的。而且更关键的是：跟踪器是带状态的（保存目标历史轨迹），跨进程共享同一个实例不仅序列化失败，还会导致多进程同时修改状态，引发数据混乱。

解决方案 #

下面给你几个可行的修复思路，按推荐优先级排序：

1. 给每个进程单独初始化跟踪器（最优解）

不要把跟踪器实例传给子进程，而是让每个子进程启动时自己创建专属的跟踪器、检测器等实例。可以用Pool的initializer和initargs参数实现：

首先，定义一个进程初始化函数，让每个子进程加载自己的资源：

def init_worker(detector, df_region, df_line, transform_matrix):
    # 定义全局变量，让子进程内的pipeline可以直接调用
    global car_tracker, ped_tracker, det_inst, df_reg, df_lin, matrix
    car_tracker = Sort_Algorithm.Sort()
    ped_tracker = Sort_Algorithm.Sort()
    det_inst = detector
    df_reg = df_region
    df_lin = df_line
    matrix = transform_matrix

然后修改pipeline函数，不再需要接收跟踪器等参数，直接用子进程内的全局变量：

def pipeline(frame, dst):
    # 这里直接使用进程内初始化好的car_tracker、ped_tracker等资源
    # 你的原有处理逻辑（检测、跟踪、画框等）
    processed_frame = ... # 替换成你实际的处理代码
    return processed_frame

最后修改主函数的进程池创建和任务提交逻辑：

if __name__ == "__main__": 
    video_name = '2016-11-18_07-30-01.h264' 
    cap = cv2.VideoCapture(video_name) 
    det = detector.CarDetector() 
    df_region, df_line = load_filter() 
    H = [-2.01134074616, -16.6502442427, -1314.05715739, -3.35391526592, -22.3546973012, 2683.63584335, -0.00130731963137, -0.0396207582264, 1] 
    matrix = np.reshape(H, (3, 3)) 

    # 创建进程池时，传入初始化函数和全局资源参数
    pool = Pool(processes=2, initializer=init_worker, initargs=(det, df_region, df_line, matrix)) 
    pending = Queue() 
    threaded_mode = True 

    while True: 
        # 处理已完成的任务
        while pending.qsize() > 0: 
            async_result = pending.get()
            # 注意：apply_async返回的是AsyncResult对象，需要调用get()拿结果
            processed_frame = async_result.get()
            cv2.imshow('video ', processed_frame) 
        
        if pending.qsize() < 2: 
            ret, frame = cap.read() 
            if not ret:
                break  # 视频读取完毕，退出循环
            
            dst = cv2.warpPerspective(frame.copy(), matrix, (frame.shape[1], frame.shape[0])) 
            if threaded_mode: 
                # 现在只需要传递frame和dst给pipeline
                task = pool.apply_async(pipeline, (frame.copy(), dst)) 
                cv2.imshow('dst', dst) 
            else: 
                # 单线程模式下，单独创建跟踪器（避免和多进程的实例冲突）
                car_tracker_single = Sort_Algorithm.Sort()
                ped_tracker_single = Sort_Algorithm.Sort()
                processed_frame = pipeline(frame.copy(), car_tracker_single, ped_tracker_single, df_region, df_line, det, dst, matrix)
                task = DummyTask(processed_frame) 
            pending.put(task) 
        
        ch = cv2.waitKey(1) 
        if ch == ord(' '): 
            threaded_mode = not threaded_mode 
        if ch == 27: 
            break 

    # 清理资源
    pool.close()
    pool.join()
    cap.release()
    cv2.destroyAllWindows()

2. 改用线程池替代进程池

如果你的任务不是纯CPU密集型（比如OpenCV内部已经做了多线程优化），可以用multiprocessing.dummy.Pool（这是基于线程的池，线程间共享内存，不需要pickle复杂实例）：

只需要修改Pool的导入语句：

from multiprocessing.dummy import Pool

其余代码基本不用改，但要注意：Sort算法必须是线程安全的（如果内部有锁，要确保锁的粒度正确），否则多线程同时访问跟踪器会导致状态异常。

3. 修改Sort算法移除不可pickle的锁

如果Sort是你自己实现的，检查内部是否用到了threading.Lock或类似的锁对象。如果锁不是必须的，直接移除；如果必须保留，把锁的初始化逻辑放到每个进程/线程内部（比如前面的initializer方式），避免跨进程传递带锁的实例。

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