FastAPI+Picamera2 MJPEG流浏览器刷新卡顿问题求助

我基于FastAPI搭建了一个服务器，用Picamera2在树莓派上推送MJPEG流。服务能正常运行，但多次刷新浏览器访问/mjpeg路径时会出现卡顿。而Picamera2官方的MJPEG服务器示例可以无限制刷新正常运行，服务器用的是默认配置的Uvicorn。

我的代码如下：

import io
import os
from threading import Condition
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from picamera2 import Picamera2
from picamera2.encoders import JpegEncoder
from picamera2.outputs import FileOutput
from fastapi.responses import HTMLResponse, StreamingResponse


app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins="http://localhost:8000",
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class StreamingOutput(io.BufferedIOBase):
    def __init__(self):
        self.frame = None
        self.condition = Condition()

    def write(self, buf):
        with self.condition:
            self.frame = buf
            self.condition.notify_all()


picam2 = Picamera2()
picam2.configure(picam2.create_video_configuration(main={"size": (640, 480)}))
output = StreamingOutput()
picam2.start_recording(JpegEncoder(), FileOutput(output))


def get_frame():
    try:
        while True:
            with output.condition:
                output.condition.wait()
                frame = output.frame
                yield (
                    b"--frame\r\n" b"Content-Type: image/jpeg\r\n\r\n" + frame + b"\r\n"
                )
    except Exception as e:
        print("Error! Frames")


@app.get("/mjpeg", response_class=StreamingResponse)
def mjpeg(request: Request):
    try:
        frames = get_frame()
        response = StreamingResponse(
            frames,
            headers={
                "Cache-Control": "no-cache, private",
                "Pragma": "no-cache",
            },
            media_type="multipart/x-mixed-replace; boundary=frame",
        )
        return response
    except Exception as e:
        print("Error! Route")

问题分析 #

卡顿的核心原因有两点：

1. 未处理客户端断开连接的情况：每次刷新浏览器都会创建新的请求连接，旧连接断开后，对应的get_frame生成器没有停止，仍然在等待output.condition的通知，导致大量僵尸线程占用资源，每次notify_all都会唤醒所有无效线程，消耗CPU资源。
2. 共享帧的竞争干扰：所有连接共享同一个StreamingOutput实例，新帧到来时notify_all会唤醒所有等待的生成器，多个连接同时读取同一帧，加上无效线程的干扰，导致有效连接的帧推送延迟。

解决方案 #

修改代码，加入客户端断开检测，确保连接断开后生成器及时退出，释放资源；同时优化并发处理配置：

import io
from threading import Condition
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from picamera2 import Picamera2
from picamera2.encoders import JpegEncoder
from picamera2.outputs import FileOutput
from fastapi.responses import StreamingResponse


app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins="http://localhost:8000",
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class StreamingOutput(io.BufferedIOBase):
    def __init__(self):
        self.frame = None
        self.condition = Condition()

    def write(self, buf):
        with self.condition:
            self.frame = buf
            self.condition.notify_all()


picam2 = Picamera2()
picam2.configure(picam2.create_video_configuration(main={"size": (640, 480)}))
output = StreamingOutput()
picam2.start_recording(JpegEncoder(), FileOutput(output))


async def get_frame(request: Request):
    try:
        while True:
            # 检测客户端是否断开连接，断开则立即退出循环
            if await request.is_disconnected():
                break
            with output.condition:
                output.condition.wait()
                frame = output.frame
                yield (b"--frame\r\n" b"Content-Type: image/jpeg\r\n\r\n" + frame + b"\r\n")
    except Exception as e:
        print(f"帧生成错误: {e}")


@app.get("/mjpeg", response_class=StreamingResponse)
async def mjpeg(request: Request):
    try:
        return StreamingResponse(
            get_frame(request),
            headers={
                "Cache-Control": "no-cache, private",
                "Pragma": "no-cache",
            },
            media_type="multipart/x-mixed-replace; boundary=frame",
        )
    except Exception as e:
        print(f"路由错误: {e}")

关键改动说明 #

• 将get_frame改为异步函数，利用FastAPI的request.is_disconnected()方法实时检测客户端连接状态，一旦断开就退出循环，避免无效线程持续占用资源。
• 调整Uvicorn启动参数，增加工作线程/进程数，提升并发处理能力：

  uvicorn main:app --host 0.0.0.0 --port 8000 --workers 2 --threads 4
  

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