多客户端访问Flask MJPEG服务卡顿及端口占用问题求助

场景说明 #

我有两个进程：

• frame_collector：从UDP socket采集JFIF帧，存入frame队列
• flask_service：Web服务器，根路由展示MJPEG，通过frame_consumer从队列取帧

原始代码 #

def frame_collector(frame): # process 1
  import io
  import socket
  SERV_IPV4, SERV_PORT = ('192.168.43.150', 8888)
  udpSock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
  udpSock.bind((SERV_IPV4,SERV_PORT))
  isWriting = False
  while 1:
    dataRecv, CLNT_ADDR = udpSock.recvfrom(65507)
      
    if not isWriting:
      if dataRecv[:2] == b'\xff\xd8': # Start of JPEG
        isWriting = True
        buf = io.BytesIO()

    if isWriting:
      buf.write(dataRecv)
      if dataRecv[-2:] == b'\xff\xd9': # End of JPEG
        isWriting = False
        buf.seek(0)
        frame.put(buf.read())

def flask_service(frame): # process 2
  from flask import Flask, Response
  app = Flask(__name__)

  def frame_consumer():
    while 1: yield b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + frame.get() + b'\r\n'

  @app.route('/mjpeg')
  def mjpeg():
    return Response(frame_consumer(),mimetype='multipart/x-mixed-replace; boundary=frame')

  @app.route('/')
  def index():
    return """
    <body style="background: blue;">
      <div style="width: 800px; margin: 0px auto;">
        <img src="/mjpeg">
      </div>
    </body>
    """

  app.run(host='192.168.43.150', threaded=True)

if __name__ == '__main__':
  from multiprocessing import Process, Queue
  from time import sleep

  frame = Queue()

  p1 = Process(target=frame_collector, args=(frame,))
  p2 = Process(target=flask_service, args=(frame,))

  p1.start()
  p2.start()
  while 1:
    try:
      sleep(5)
    except:
      break
  p1.terminate()
  p2.terminate()
  print('All processes terminated!')

问题现象 #

单个客户端访问时MJPEG流畅（60FPS），但多客户端同时访问时卡顿，客户端越多越严重，怀疑是frame这个跨线程共享的Queue导致的问题。

测试反馈（更新内容） #

尝试了他人提供的代码后，先修复了flask_service缺参数的问题，服务器能正常运行，第一个客户端访问/mjpeg时保持60FPS，但新客户端访问时陷入加载状态，同时出现错误：

Traceback (most recent call last):
  File "C:\Users\User\AppData\Local\Programs\Python\Python39\lib\multiprocessing\process.py", line 315, in _bootstrap
    self.run()
  File "C:\Users\User\AppData\Local\Programs\Python\Python39\lib\multiprocessing\process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "c:\Users\User\Documents\PlatformIO\Projects\udp-wifi-camera\src\test_punya_orang.py", line 9, in frame_collector
    udpSock.bind((SERV_IPV4,SERV_PORT))
OSError: [WinError 10048] Only one usage of each socket address (protocol/network address/port) is normally permitted
Ending up.

错误原因是新客户端访问时尝试创建绑定相同地址的socket，导致端口被占用。

解决方案 #

核心问题分析 #

1. 原始代码中Queue.get()是阻塞的，每个客户端的线程都会从队列取帧，队列中的帧被一个客户端取走后，其他客户端只能等待新帧，这是卡顿的根源——单帧无法被多客户端共享。
2. 测试时的端口占用问题，是因为错误地在客户端连接时重复创建frame_collector进程，导致重复绑定UDP端口。

修复方案 #

1. 实现帧的多客户端共享：用共享变量存储最新帧

放弃Queue传递帧，改用多进程安全的共享变量（multiprocessing.Array+multiprocessing.Value）存储最新采集的JPEG帧，所有客户端都能读取同一帧，避免队列的“抢帧”问题。

2. 确保UDP采集进程仅启动一次

frame_collector作为独立进程仅启动一次，绑定UDP端口后持续采集，避免重复绑定导致端口占用。

完整修复代码 #

from multiprocessing import Process, Array
from time import sleep
import io
import socket
from flask import Flask, Response

# 多进程共享的帧存储：用Array存储字节数据，Value存储当前帧长度
def frame_collector(frame_buf, frame_len):
    SERV_IPV4, SERV_PORT = ('192.168.43.150', 8888)
    udpSock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    udpSock.bind((SERV_IPV4, SERV_PORT))
    isWriting = False
    buf = io.BytesIO()

    while True:
        dataRecv, CLNT_ADDR = udpSock.recvfrom(65507)
        
        if not isWriting:
            if dataRecv[:2] == b'\xff\xd8':  # JPEG开始标记
                isWriting = True
                buf = io.BytesIO()
                buf.write(dataRecv)
        else:
            buf.write(dataRecv)
            if dataRecv[-2:] == b'\xff\xd9':  # JPEG结束标记
                isWriting = False
                frame_data = buf.getvalue()
                # 加锁更新共享数据，保证读写安全
                with frame_len.get_lock():
                    frame_len.value = len(frame_data)
                if len(frame_data) <= len(frame_buf):
                    frame_buf[:len(frame_data)] = frame_data

def flask_service(frame_buf, frame_len):
    app = Flask(__name__)

    def frame_consumer():
        while True:
            # 读取共享帧数据
            with frame_len.get_lock():
                current_len = frame_len.value
            if current_len > 0 and current_len <= len(frame_buf):
                frame_data = bytes(frame_buf[:current_len])
                yield b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + frame_data + b'\r\n'

    @app.route('/mjpeg')
    def mjpeg():
        return Response(frame_consumer(), mimetype='multipart/x-mixed-replace; boundary=frame')

    @app.route('/')
    def index():
        return """
        <body style="background: blue;">
          <div style="width: 800px; margin: 0px auto;">
            <img src="/mjpeg">
          </div>
        </body>
        """

    app.run(host='192.168.43.150', threaded=True)

if __name__ == '__main__':
    # 定义共享缓冲区：假设最大帧长度为1MB（可根据实际调整）
    MAX_FRAME_SIZE = 1024 * 1024
    frame_buf = Array('b', MAX_FRAME_SIZE)
    frame_len = Array('i', 1)  # 存储当前帧的实际长度

    p1 = Process(target=frame_collector, args=(frame_buf, frame_len))
    p2 = Process(target=flask_service, args=(frame_buf, frame_len))

    p1.start()
    p2.start()

    try:
        while True:
            sleep(5)
    except KeyboardInterrupt:
        p1.terminate()
        p2.terminate()
        print('All processes terminated!')

修复说明 #

• 共享帧存储：用multiprocessing.Array创建字节缓冲区，frame_len记录当前帧的实际长度，加锁保证读写安全，所有客户端都能读取最新帧，避免队列阻塞抢帧。
• UDP进程单实例：frame_collector仅启动一次，绑定UDP端口后持续采集，不会重复绑定导致端口占用错误。
• Flask线程安全：Flask开启threaded=True后，每个客户端请求对应独立线程，各线程读取共享帧互不干扰，保证多客户端流畅播放。

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