基于FastAPI与OpenCV的RTMP流API处理问题求助

1. 先验证RTMP流的读取能力 #

OpenCV对RTMP的支持依赖FFmpeg，先单独确认流能否被正常读取：

import cv2
import numpy as np

cap = cv2.VideoCapture("rtmp://62.113.210.250/medienasa-live/rbw_high")
if not cap.isOpened():
    print("无法打开RTMP流，请检查FFmpeg是否正确安装并被OpenCV调用")
else:
    for _ in range(10):
        ret, frame = cap.read()
        if not ret or frame is None:
            print("读取帧失败")
            break
        print(f"读取到帧，尺寸: {frame.shape}")
    cap.release()

如果这段脚本报错，说明是环境问题：

• 卸载现有opencv-python，安装带FFmpeg的版本：pip install opencv-contrib-python-headless
• 确保系统已安装FFmpeg（Ubuntu用apt install ffmpeg，CentOS用yum install ffmpeg）

2. 修复接口的异常处理逻辑 #

原代码缺少错误捕获，任何环节出错都会导致生成器静默卡住，修改后增加流重连、异常捕获：

@app.post("/stream_detect")
def stream_detect(link: str = Form()):
    def get_capture():
        cap = cv2.VideoCapture(link)
        return cap if cap.isOpened() else None

    capture = get_capture()
    if not capture:
        return {"error": "无法初始化视频流"}
    
    def detect(stream):
        while True:
            try:
                err, frame = stream.read()
                if not err or frame is None:
                    # 流断开，尝试重新连接
                    stream.release()
                    stream = get_capture()
                    if not stream:
                        # 重连失败，输出占位图
                        placeholder = np.zeros((480, 640, 3), dtype=np.uint8)
                        flag, encoded = cv2.imencode(".jpg", placeholder)
                        yield b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + encoded.tobytes() + b'\r\n'
                        continue
                    continue
                
                # 模型推理环节
                detection = model(frame)
                detection.render()
                
                (flag, encodedimage) = cv2.imencode(".jpg", detection.imgs[0])
                if not flag:
                    continue
                
                yield b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + bytearray(encodedimage) + b'\r\n'
            
            except Exception as e:
                print(f"帧处理错误: {str(e)}")
                continue
    
    return StreamingResponse(detect(capture), media_type="multipart/x-mixed-replace;boundary=frame")

3. 分步调试定位问题 #

• 先剥离模型推理：写一个仅读取流并输出的测试接口，确认基础流输出正常：

  @app.post("/test_raw_stream")
  def test_raw_stream(link: str = Form()):
      capture = cv2.VideoCapture(link)
      if not capture.isOpened():
          return {"error": "无法打开流"}
      
      def generate():
          while True:
              ret, frame = capture.read()
              if not ret:
                  break
              flag, encoded = cv2.imencode('.jpg', frame)
              if not flag:
                  continue
              yield b'--frame\r\nContent-Type: image/jpeg\r\n\r\n' + encoded.tobytes() + b'\r\n'
      
      return StreamingResponse(generate(), media_type="multipart/x-mixed-replace;boundary=frame")
  

  如果这个接口能正常输出流，说明问题出在模型推理环节；如果仍无输出，排查流读取或FastAPI响应配置。

• 用curl验证接口输出：避免浏览器兼容性问题，直接用命令行测试：

  curl -X POST -F "link=rtmp://62.113.210.250/medienasa-live/rbw_high" http://18.231.43.242:8000/stream_detect --output -
  

  如果能看到--frame开头的二进制数据，说明接口在输出，问题在浏览器端；如果无输出，说明接口未生成数据。

4. 排查性能瓶颈 #

模型推理耗时过长会导致流延迟极高，看起来像无输出，加计时代码查看单帧耗时：

import time

# 在模型推理前添加
start = time.time()
detection = model(frame)
print(f"单帧推理耗时: {time.time() - start:.2f}s")

如果耗时超过0.1s（10fps），需要优化模型：

• 用TensorRT/ONNX Runtime做模型量化加速
• 降低输入帧尺寸（比如把帧缩放到640x480再推理）

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