基于GStreamer Python绑定搭建RTSP与录像管道遇冻结问题求助

核心问题成因 #

1. 多客户端信号绑定冲突：每次有RTSP客户端连接时，都会重复给同一个appsink绑定new-sample回调，导致同一份媒体样本被多个appsrc争抢推送，造成流时序混乱、冻结。
2. AppSrc格式协商失败：RTSP管道中的appsrc未明确指定媒体格式，GStreamer自动协商可能出错，客户端无法持续解码。
3. 分支处理速度失衡：主管道的tee分支未加缓冲队列，文件写入的慢处理可能拖慢RTSP流的传输节奏。

修复后的完整代码 #

import gi
import signal
import sys
from queue import Queue
from threading import Lock

gi.require_version('Gst', '1.0')
gi.require_version('GstRtspServer', '1.0')
from gi.repository import Gst, GstRtspServer, GLib

Gst.init(None)

# 线程安全队列，用于主管道与RTSP管道间的媒体样本传递
sample_queue = Queue(maxsize=30)
queue_lock = Lock()

def signal_handler(sig, frame):
    print("捕获到信号，清理资源并停止主循环。")
    mainloop.quit()

def on_new_sample(appsink):
    sample = appsink.emit("pull-sample")
    if sample:
        with queue_lock:
            if not sample_queue.full():
                sample_queue.put(sample)
    return Gst.FlowReturn.OK

def push_sample_to_appsrc(appsrc):
    with queue_lock:
        if not sample_queue.empty():
            sample = sample_queue.get()
            appsrc.emit("push-sample", sample)
    return GLib.SOURCE_CONTINUE

BITRATE = 2000000

main_pipeline_str = f"""
v4l2src device=/dev/video7 do-timestamp=true ! video/x-raw, format=YUY2, width=1920, height=1080, framerate=30/1 ! nvvidconv ! video/x-raw(memory:NVMM) ! nvv4l2h264enc bitrate={BITRATE} preset-level=4 insert-sps-pps=true ! tee name=t
t. ! queue ! h264parse ! splitmuxsink location=/external/video_%05d.mkv max-size-time=100000000000 max-files=10
t. ! queue ! h264parse ! appsink name=appsink emit-signals=true sync=false
"""

main_pipeline = Gst.parse_launch(main_pipeline_str)
main_pipeline.set_state(Gst.State.PLAYING)
appsink = main_pipeline.get_by_name("appsink")
# 仅绑定一次信号，避免多客户端冲突
appsink.connect("new-sample", on_new_sample)

def error_cb(bus, msg):
    err, debug_info = msg.parse_error()
    print(f"错误: {err.message}")

bus = main_pipeline.get_bus()
bus.add_signal_watch()
bus.connect("message::error", error_cb)

class TestRtspMediaFactory(GstRtspServer.RTSPMediaFactory):
    def __init__(self):
        GstRtspServer.RTSPMediaFactory.__init__(self)
        self.set_shared(True)

    def do_create_element(self, url):
        print("新客户端连接！")
        # 为appsrc指定明确的H264格式，匹配主管道输出
        pipeline_str = """
            appsrc name=appsrc is-live=true do-timestamp=true caps="video/x-h264, stream-format=byte-stream, alignment=au" ! 
            h264parse ! rtph264pay config-interval=1 pt=96 name=pay0
        """
        pipeline = Gst.parse_launch(pipeline_str)
        appsrc = pipeline.get_by_name("appsrc")
        # 定时从队列取样本推送到appsrc，间隔与帧率匹配
        GLib.timeout_add(30, push_sample_to_appsrc, appsrc)
        return pipeline

rtsp_server = GstRtspServer.RTSPServer()
rtsp_server.set_service("8557")
factory = TestRtspMediaFactory()
mount_points = rtsp_server.get_mount_points()
mount_points.add_factory("/camera", factory)
rtsp_server.attach(None)

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

mainloop = GLib.MainLoop()
mainloop.run()

关键修复说明 #

• 线程安全队列解耦：主管道的appsink只负责向队列写入样本，每个RTSP客户端的appsrc通过定时回调读取样本，彻底避免多客户端的资源竞争。
• 明确媒体格式：给RTSP管道的appsrc指定与主管道一致的H264格式参数，确保格式协商一次成功，消除解码中断隐患。
• 添加缓冲队列：在tee的两个分支都加入queue元素，平衡文件写入与RTSP流的处理速度，防止单一分支阻塞导致整体管道卡顿。
• 单次信号绑定：仅在主管道初始化时绑定new-sample信号，客户端连接时不再重复绑定，消除回调冲突。

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