Python：RTSP流故障时如何检测脚本挂停并自动重启？

问题 #

我在做RTSP视频流拉流分析，代码短时间运行正常，但长时间拉流后会挂停。Wireshark抓包显示：服务器（10.61.41.4）发送Keep-alive，客户端回复ACK，但服务器不再返回数据，客户端发重置报文引发TCP重传。推测是服务器端流应用停止，但我无法控制服务器；就算服务器重启，脚本还是会挂着。

我想找代码层面的方法，检测脚本挂停超过30秒后自动重启整个脚本，这比检测TCP重传次数（超过4次）更简便。

之前试过用ffprobe.exe，本来以为能检测流是否存活，但流断开时它没有任何返回。选ffprobe是因为它比opencv的cv2.VideoCapture()超时响应更快，但只在脚本启动时有效，对付不了长时间运行后的挂停场景。怀疑是命令参数用错，或者应该换其他方法？

附上测试的ffprobe代码：

def probe_stream(path, cam_name):
     
    cmnd = [r'C:\ffmpeg-2022\bin\ffprobe.exe', '-rtsp_transport', 'tcp', '-show_format', 
    '-pretty', '-loglevel', 'quiet', '-of', 'json', path]
    p = subprocess.Popen(cmnd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    print(p)
    print(p.communicate())
    s = p.communicate()[0]
    print(s)

    if p.returncode == 0:
        out, err = p.communicate()
        p.wait()
        probe_dct = json.loads(out)

        try:
            if probe_dct:
                return True
            elif err:
                print(err)
        except Exception as e:
            print(e)

解决方案 #

一、修复ffprobe的流存活检测逻辑 #

你的ffprobe代码存在3个核心问题：

1. 多次调用p.communicate()：第一次调用已读完子进程输出，后续调用会返回空值
2. 未设置超时：流挂停时ffprobe会持续阻塞，不会主动退出
3. 参数适配错误：-show_format仅读取流格式信息，适合启动前校验，无法持续监听运行中的流状态

修改后的ffprobe监控代码，通过限时读取流数据判断存活状态：

import subprocess
import json

def check_rtsp_alive(path, timeout=30):
    # 配置ffprobe读取5秒流数据，超时判定为挂停
    cmnd = [
        r'C:\ffmpeg-2022\bin\ffprobe.exe',
        '-rtsp_transport', 'tcp',
        '-show_streams',
        '-loglevel', 'error',
        '-read_intervals', '%+5',  # 仅读取5秒流数据
        '-of', 'json',
        path
    ]
    try:
        # 设置超时，超过指定时间无返回则判定流挂停
        result = subprocess.run(cmnd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout)
        if result.returncode == 0:
            json.loads(result.stdout)
            return True
        else:
            print(f"流异常: {result.stderr.decode('utf-8')}")
            return False
    except subprocess.TimeoutExpired:
        print("流无响应，判定为挂停")
        return False
    except Exception as e:
        print(f"检测出错: {str(e)}")
        return False

二、脚本挂停检测与自动重启方案 #

方法1：主进程监控子进程心跳

将拉流分析的核心逻辑封装为独立函数，主进程启动子进程运行该逻辑，同时定期检查流状态，超过30秒无响应则重启子进程：

import subprocess
import time
import sys

def main_stream_analysis():
    # 此处放置你的RTSP拉流、分析核心代码
    # 例如用ffmpeg/opencv实现帧处理逻辑
    pass

def monitor_and_restart():
    rtsp_path = "rtsp://10.61.41.4/your-stream-path"
    while True:
        # 启动子进程运行拉流逻辑
        proc = subprocess.Popen([sys.executable, __file__, "--run-analysis"])
        last_alive_time = time.time()

        while proc.poll() is None:
            # 每5秒检查一次流状态
            if check_rtsp_alive(rtsp_path):
                last_alive_time = time.time()
            # 超过30秒无响应则杀死子进程并重启
            elif time.time() - last_alive_time > 30:
                print("脚本挂停超过30秒，重启中...")
                proc.kill()
                proc.wait()
                break
            time.sleep(5)
        
        print("子进程退出，准备重启...")
        time.sleep(2)

if __name__ == "__main__":
    if len(sys.argv) > 1 and sys.argv[1] == "--run-analysis":
        main_stream_analysis()
    else:
        monitor_and_restart()

方法2：ffmpeg拉流时内置超时

直接在ffmpeg拉流命令中设置超时参数，当流无数据时ffmpeg自动退出，脚本捕获退出状态后重启：

import subprocess
import time

def pull_and_analyze_rtsp(path):
    cmnd = [
        r'C:\ffmpeg-2022\bin\ffmpeg.exe',
        '-rtsp_transport', 'tcp',
        '-stimeout', '30000000',  # 30秒超时（单位：微秒）
        '-i', path,
        '-c:v', 'copy',
        '-f', 'null', '-'  # 仅拉流不输出文件
    ]
    try:
        proc = subprocess.run(cmnd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        return proc.returncode == 0
    except Exception as e:
        print(f"拉流出错: {str(e)}")
        return False

def auto_restart_loop(path):
    while True:
        print("启动拉流分析...")
        success = pull_and_analyze_rtsp(path)
        if not success:
            print("拉流挂停/异常，10秒后重启...")
            time.sleep(10)

关键注意事项 #

• 避免用cv2.VideoCapture()做长时间监控，其超时机制薄弱，极易阻塞
• -stimeout参数仅对TCP传输有效（你已指定-rtsp_transport tcp），UDP传输需改用其他参数
• 监控频率建议设为5-10秒一次，避免占用过多系统资源

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