如何用Python实现函数并行执行？优化帧处理串行流程

嗨，这个场景我太熟悉了！原来的串行循环把获取、处理、显示串在一起，每一步都得等上一步完才能继续，效率确实拉胯。要实现process_frame()运行时同步执行get_frame()，核心就是把这两个任务拆到不同的执行流里跑，用一个队列来做帧数据的缓冲和传递，这样处理当前帧的同时，下一个帧已经在后台默默获取好了。下面给你两种实用的Python实现方案：

这种方案适合get_frame()是IO密集型的场景（比如读取摄像头、拉取网络视频流），线程开销小，实现起来也直观。

思路是：开一个后台线程专门负责持续获取帧，把帧放到队列里；主线程专注于从队列取帧、处理、显示。两者独立运行，互不阻塞。

import threading
import queue
import cv2  # 这里用OpenCV做示例，你可以替换成自己的帧获取/处理逻辑

# 帧队列：缓冲获取到的帧，设置最大长度避免内存溢出
frame_queue = queue.Queue(maxsize=2)
# 停止信号：用来优雅终止后台线程，避免强制退出导致资源泄漏
stop_event = threading.Event()

def frame_producer():
    """后台线程：持续获取帧并放入队列"""
    # 初始化视频源（这里是本地摄像头，换成你的视频路径/网络流地址都行）
    cap = cv2.VideoCapture(0)
    
    while not stop_event.is_set():
        ret, frame = cap.read()
        if not ret:
            # 读取失败（比如摄像头断开），直接退出循环
            break
        
        # 如果队列满了，丢弃最旧的帧（避免阻塞获取线程）
        if frame_queue.full():
            try:
                frame_queue.get_nowait()
            except queue.Empty:
                pass
        frame_queue.put(frame)
    
    # 释放资源
    cap.release()

def main():
    # 启动帧获取线程
    producer_thread = threading.Thread(target=frame_producer)
    producer_thread.start()
    
    while not stop_event.is_set():
        try:
            # 从队列取帧，设置超时时间防止主线程无限阻塞
            frame = frame_queue.get(timeout=1)
        except queue.Empty:
            continue
        
        # 执行帧处理逻辑（替换成你的process_frame函数）
        processed_frame = process_frame(frame)
        # 显示处理后的帧（替换成你的show_frame函数）
        show_frame(processed_frame)
        
        # 按Q键退出程序
        if cv2.waitKey(1) & 0xFF == ord('q'):
            stop_event.set()
            break
    
    # 等待后台线程结束，释放资源
    producer_thread.join()
    cv2.destroyAllWindows()

# 示例：自定义帧处理函数
def process_frame(frame):
    # 比如把彩色帧转成灰度帧
    return cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

# 示例：自定义帧显示函数
def show_frame(frame):
    cv2.imshow('Processed Frame', frame)

if __name__ == "__main__":
    main()

如果你不想手动管理线程，用ThreadPoolExecutor可以更优雅地实现，它会帮你处理线程的创建和销毁。

import concurrent.futures
import cv2
import queue

frame_queue = queue.Queue(maxsize=2)
stop_flag = False

def get_frame():
    """持续获取帧并放入队列"""
    cap = cv2.VideoCapture(0)
    global stop_flag
    
    while not stop_flag:
        ret, frame = cap.read()
        if ret:
            if frame_queue.full():
                try:
                    frame_queue.get_nowait()
                except queue.Empty:
                    pass
            frame_queue.put(frame)
    
    cap.release()

def main():
    global stop_flag
    # 用线程池管理帧获取任务（max_workers设为1就够了）
    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        executor.submit(get_frame)
        
        while not stop_flag:
            try:
                frame = frame_queue.get(timeout=1)
            except queue.Empty:
                continue
            
            processed = process_frame(frame)
            show_frame(processed)
            
            if cv2.waitKey(1) & 0xFF == ord('q'):
                stop_flag = True
                break
    
    cv2.destroyAllWindows()

# 同样的示例处理/显示函数
def process_frame(frame):
    return cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)

def show_frame(frame):
    cv2.imshow('Pool Processed Frame', frame)

if __name__ == "__main__":
    main()

• 队列大小设置：maxsize要根据你的内存和处理速度调整，太大占内存，太小容易丢帧。如果对实时性要求高，建议设为1或2。
• CPU密集型处理：如果process_frame()是CPU密集型任务（比如复杂的图像AI推理），线程可能发挥不了并行优势，这时候可以换成multiprocessing模块（进程并行），注意要用multiprocessing.Queue来传递帧数据。
• 优雅终止：一定要用Event或全局变量来控制退出，避免直接强制终止导致摄像头/文件句柄泄漏。
• 线程安全：大部分图像处理库（比如OpenCV、PIL）都是线程安全的，但如果用了自定义的资源共享逻辑，要注意加锁（threading.Lock）。

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