GStreamer多摄像头RTP流动态添加解复用元素链失败问题求助

核心问题点 #

你的代码无法正常运行的核心原因是动态添加的元素没有同步设置为PLAYING状态，GStreamer管道启动后新加入的元素默认处于NULL状态，无法处理数据，同时还有几个细节问题需要修正：

• 你先把管道切到PLAYING状态才监听pad-added信号，存在信号触发早于监听绑定的竞态问题，应该先绑定信号再启动管道
• 用time.sleep会阻塞GLib主循环，GStreamer的信号回调依赖主循环运行，阻塞后会导致回调无法触发
• 不需要处理的RTCP流可以直接过滤，避免不必要的资源占用

修正后的完整代码 #

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


class Video():
    def __init__(self, port=5000):
        Gst.init(None)
        self.port = port
        # UDP视频复用流监听端口5000
        self.launch_pipeline = [
            'rtpssrcdemux name=rtpdemux',
            f'udpsrc name=udpsrc port={port}',
            '! rtpdemux.sink'
        ]
        self.start_gst(self.launch_pipeline)

    def start_gst(self, config):
        command = ' '.join(config)
        print("启动管道:", command)
        self.video_pipe = Gst.parse_launch(command)
        # 先绑定所有信号再启动管道，避免信号丢失
        self.demuxer = self.video_pipe.get_by_name('rtpdemux')
        self.demuxer.connect("pad-added", self._demuxer_new_pad)
        bus = self.video_pipe.get_bus()
        bus.add_signal_watch()
        bus.connect("message::error", self._on_error)
        bus.connect("message::eos", self._on_eos)
        self.app_sink = {}
        # 信号绑定完成后再将管道切到PLAYING状态
        self.video_pipe.set_state(Gst.State.PLAYING)

    def _on_error(self, _, message):
        err, debug = message.parse_error()
        print("错误: %s" % err, debug)

    def _on_eos(self, _, message):
        print("流结束!")

    def _demuxer_new_pad(self, demuxer, pad):
        name = pad.get_name()
        print(f"收到新Pad: {name}")
        # 不需要处理RTCP流直接跳过
        if name.startswith("rtcp"):
            return

        # 创建后续处理链路元素
        depay = Gst.ElementFactory.make("rtpvrawdepay", f"depay_{name}")
        convert = Gst.ElementFactory.make("videoconvert", f"convert_{name}")
        sink = Gst.ElementFactory.make("appsink", f"appsink_{name}")
        sink.set_property("emit-signals", True)
        sink.set_property("sync", False)
        sink.set_property("max-buffers", 0)
        sink.set_property("drop", True)
        sink.connect('new-sample', self._new_frame)

        # 将元素添加到管道
        self.video_pipe.add(depay)
        self.video_pipe.add(convert)
        self.video_pipe.add(sink)

        # 同步新元素状态和管道保持一致，核心修复点
        depay.sync_state_with_parent()
        convert.sync_state_with_parent()
        sink.sync_state_with_parent()

        # 连接链路
        depay.link(convert)
        convert.link(sink)
        chain_pad = depay.get_static_pad("sink")
        pad.link(chain_pad)
        self.app_sink[name] = sink

    def _new_frame(self, sink):
        print("收到新视频帧")
        sample = sink.emit('pull-sample')
        caps = sample.get_caps()
        name = sink.get_name()
        print(f"帧格式: {caps.to_string()}, 对应SSRC: {name.split('_')[-1]}")
        return Gst.FlowReturn.OK

if __name__ == '__main__':
    video = Video(port=5000)
    # 用GLib主循环代替sleep，保证信号回调正常触发
    loop = GLib.MainLoop()
    loop.run()

后续切换H.264适配说明 #

后续你切换到H.264格式时，只需要把代码中的rtpvrawdepay替换为rtph264depay，再在解封装后增加对应GPU解码元件即可：N卡用nvh264dec、Intel核显用vaapih264dec、AMD显卡用vah264dec。

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