多进程环境下Flask获取其他进程视频帧的实现问题

核心问题在于app.run()会阻塞主进程，导致主进程无法主动从多进程队列取帧更新self.frame。解决方案是让Flask的视频流生成器直接从队列获取最新帧，同时将Flask服务放在子线程中运行，避免阻塞主进程。

修正并优化后的Flask代码 #

from flask import Flask, render_template, Response
import threading

class FlaskAppWrapper(object):
    def __init__(self, app, frame_queue, **configs):
        self.app = app
        self.frame_queue = frame_queue  # 接收多进程队列
        self.configs(**configs)
        self.frame = None

    def configs(self, **configs):
        # 修正字典遍历错误，用items()获取键值对
        for config, value in configs.items():
            self.app.config[config.upper()] = value

    def add_endpoint(self, endpoint=None, endpoint_name=None, handler=None, methods=['GET'], *args, **kwargs):
        self.app.add_url_rule(endpoint, endpoint_name, handler, methods=methods, *args, **kwargs)
    
    def run(self, **kwargs):
        self.app.run(**kwargs)

    # 修正方法缺少self参数的问题
    def index(self):
        return render_template('index.html')

    def gen(self):
        while True:
            # 清空队列旧帧，只保留最新一帧，避免视频延迟
            while not self.frame_queue.empty():
                try:
                    self.frame = self.frame_queue.get(timeout=0.1)
                except:
                    break
            # 确保有帧再返回
            if self.frame is not None:
                yield (b'--frame\r\n'
                       b'Content-Type: image/jpeg\r\n\r\n' + self.frame + b'\r\n\r\n')
            
    def video_feed(self):
        return Response(self.gen(),
                        mimetype='multipart/x-mixed-replace; boundary=frame')

def run_webserver(frame_queue):
    flask_app = Flask(__name__)
    app = FlaskAppWrapper(flask_app, frame_queue)

    app.add_endpoint('/index', 'index', app.index, methods=['GET'])
    app.add_endpoint('/video_feed', 'video_feed', app.video_feed, methods=['GET'])

    # debug模式需关闭重载，避免多进程队列冲突
    app.run(host='0.0.0.0', port=5000, debug=False, use_reloader=False)

主进程调用示例 #

主进程将Flask放在子线程运行，同时启动视频采集进程：

from multiprocessing import Queue, Process
import time
import cv2
import numpy as np

# 模拟视频采集进程：往队列推送帧
def video_capture_process(frame_queue):
    while True:
        # 实际项目中替换为真实视频帧采集逻辑
        frame = np.zeros((480, 640, 3), dtype=np.uint8)
        _, jpeg_frame = cv2.imencode('.jpg', frame)
        frame_queue.put(jpeg_frame.tobytes())
        time.sleep(0.03)  # 模拟30fps

if __name__ == '__main__':
    # 创建带长度限制的队列，避免内存溢出
    frame_queue = Queue(maxsize=10)

    # 启动视频采集进程
    capture_proc = Process(target=video_capture_process, args=(frame_queue,))
    capture_proc.start()

    # 子线程启动Flask，不阻塞主进程
    web_thread = threading.Thread(target=run_webserver, args=(frame_queue,))
    web_thread.daemon = True  # 主进程退出时自动终止线程
    web_thread.start()

    # 主进程可继续执行其他任务
    while True:
        time.sleep(1)

关键说明 #

• 队列帧处理：生成器每次循环会清空队列旧帧，只取最新帧，避免视频流出现延迟。
• 线程隔离：Flask服务在子线程运行，主进程可正常管理视频采集等其他进程。
• debug模式注意：若需开启debug，必须设置use_reloader=False，否则Flask会启动额外进程导致队列无法共享。
• 队列容量限制：设置队列maxsize，防止旧帧累积占用过多内存。

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