Python多进程优化带while循环与共享资源的OpenCV图像拼接任务

• 多进程全局变量不共享：Windows系统下Python的multiprocessing默认用spawn启动模式，子进程启动时会重新导入你的代码文件，不会执行if __name__ == '__main__'包裹的逻辑，因此子进程内的frames_1、frames_2都是空列表，while循环判断条件不成立直接跳过。
• 共享变量操作不安全：即使你用了支持跨进程共享的存储结构，直接多进程同时执行pop(0)、append操作也会有数据竞争问题，出现重复处理、丢失数据、结果错乱的情况。
• 多线程不适配你的场景：图像拼接是CPU密集型任务，Python全局解释器锁（GIL）会导致多线程只能使用单个CPU核心，完全达不到4倍加速的效果，必须用多进程方案。

直接使用multiprocessing.Pool进程池实现，不需要手动管理进程生命周期，也不需要维护全局共享变量，把待处理的图像对作为参数传入工作函数，处理完成后主进程统一收集结果即可，实现简单且稳定性高。

import multiprocessing
import time
import cv2

# 单组图像拼接工作函数，不要用全局变量，参数和返回值都单独处理
def stitch_single_pair(img_pair):
    img1, img2 = img_pair
    stitcher = cv2.Stitcher_create(cv2.STITCHER_PANORAMA)
    status, result = stitcher.stitch((img1, img2))
    if status == 0:
        return result
    else:
        print("image stitching failed")
        return None

# converting the video into frames for individual image processing
def convert_video_to_frames():
    frames_1 = []
    frames_2 = []
    cap = cv2.VideoCapture("Sample_video_1.mp4")
    ret = True
    while ret:
        ret, img = cap.read()
        if ret:
            frames_1.append(img)
    cap.release()

    cap = cv2.VideoCapture("Sample_video_2.mp4")
    ret = True
    while ret:
        ret, img = cap.read()
        if ret:
            frames_2.append(img)
    cap.release()

    # 保证两个视频帧数量一致，取最短的长度
    min_len = min(len(frames_1), len(frames_2))
    # 打包成待处理的图像对列表
    return list(zip(frames_1[:min_len], frames_2[:min_len]))

# converting final output images back to video
def convert_frames_to_video(panorama):
    print("now creating stitched image video")
    height, width, layers = panorama[0].shape
    size = (width, height)
    out = cv2.VideoWriter('project.avi', cv2.VideoWriter_fourcc(*'DIVX'), 15, size)
    for img in panorama:
        out.write(img)
    out.release()

if __name__ == '__main__':
    # 先读取所有视频帧，打包成待处理任务列表
    task_list = convert_video_to_frames()
    print(f"总共有{len(task_list)}组待拼接图像")

    start = time.perf_counter()
    # 开启4进程的进程池处理任务
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(stitch_single_pair, task_list)
    # 过滤掉拼接失败的None值
    panorama = [res for res in results if res is not None]
    finish = time.perf_counter()
    print(f'finished in {round(finish - start, 2)} seconds(s)')

    print("now converting images to video...")
    convert_frames_to_video(panorama)

• 进程数设置不要超过你的CPU物理核心数，否则进程切换开销会抵消多进程加速效果。
• 如果视频帧数量很大，一次性读取所有帧占内存过高，可以用multiprocessing.Queue做流式任务分发，读一批处理一批，降低内存占用。
• 可以提前检查两个视频的帧数量是否匹配，避免出现无法配对的情况。

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