Python GStreamer RTSP Server多客户端共享流水线失效问题求助

问题背景 #

已在代码中调用self.factory.set_shared(True)，但实际测试中CPU占用随客户端连接数上升，每次连接都会触发do_configure方法，未实现流水线共享效果。根据GStreamer官方文档：

通过GstRtspServer.RTSPMediaFactory.set_shared设置共享标志后，工厂的媒体可被共享。当工厂设为共享时，若URL匹配，GstRtspServer.RTSPMediaFactory.construct将返回同一个GstRtspServer.RTSPMedia实例。

原代码问题分析 #

原代码的核心问题在于：

• 每个客户端连接时，do_create_element会创建一条包含x264enc编码环节的完整流水线，即使设置了shared=True，编码操作仍会为每个客户端重复执行，导致CPU占用随客户端数量线性上升
• 通过OpenCV的VideoCapture读取帧再推给appsrc的方式，引入了额外的内存拷贝和转换开销，也不利于流水线的原生共享机制

解决方案：构建共享编码的流水线架构 #

要实现真正的资源共享，需要将视频采集、编码环节作为公共流水线，通过tee元素将编码后的流分流给多个客户端专属的RTSP打包环节，所有客户端共享同一套采集和编码流程，仅在最后阶段各自处理RTSP协议的打包工作。

修改后的完整代码 #

import gi

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

GObject.threads_init()
Gst.init(None)

class SharedSensorFactory(GstRtspServer.RTSPMediaFactory):
    def __init__(self):
        super(SharedSensorFactory, self).__init__()
        self.set_shared(True)
        # 构建公共流水线：采集→格式转换→编码→分流tee
        self.pipeline = Gst.Pipeline.new("shared-pipeline")
        
        # 采集元素
        src = Gst.ElementFactory.make("nvv4l2camerasrc", "src")
        src.set_property("device", "/dev/video0")
        
        # NVMM格式转换为I420（适配x264enc）
        conv = Gst.ElementFactory.make("nvvidconv", "conv")
        caps_filter = Gst.ElementFactory.make("capsfilter", "caps-filter")
        caps_filter.set_property("caps", Gst.Caps.from_string(
            "video/x-raw,format=I420,width=1280,height=960,framerate=25/1"
        ))
        
        # H264编码（仅执行一次，所有客户端共享）
        encoder = Gst.ElementFactory.make("x264enc", "encoder")
        encoder.set_property("speed-preset", "ultrafast")
        encoder.set_property("tune", "zerolatency")
        
        # tee元素：将编码后的流分流给多个客户端
        tee = Gst.ElementFactory.make("tee", "tee")
        tee.set_property("allow-not-linked", True)
        
        # 将公共元素添加到流水线并链接
        self.pipeline.add(src)
        self.pipeline.add(conv)
        self.pipeline.add(caps_filter)
        self.pipeline.add(encoder)
        self.pipeline.add(tee)
        
        src.link(conv)
        conv.link(caps_filter)
        caps_filter.link(encoder)
        encoder.link(tee)
        
        # 启动公共流水线
        self.pipeline.set_state(Gst.State.PLAYING)

    def do_create_element(self, url):
        # 为每个客户端创建专属分支：队列→RTSP打包
        launch_str = (
            "queue name=queue ! rtph264pay config-interval=1 name=pay0 pt=96"
        )
        return Gst.parse_launch(launch_str)

    def do_configure(self, rtsp_media):
        # 获取客户端分支的元素
        client_pipeline = rtsp_media.get_element()
        queue = client_pipeline.get_child_by_name("queue")
        
        # 获取公共流水线的tee元素
        tee = self.pipeline.get_child_by_name("tee")
        
        # 动态请求tee的输出pad并链接到客户端的队列
        tee_pad = tee.request_pad_simple("src_%u")
        queue_sink_pad = queue.get_static_pad("sink")
        tee_pad.link(queue_sink_pad)
        
        # 客户端断开时释放tee的pad，避免资源泄漏
        def on_media_unprepared(media, pad):
            pad.send_event(Gst.Event.new_eos())
            tee.release_request_pad(pad)
        
        rtsp_media.connect("unprepared", on_media_unprepared, tee_pad)

class GstServer(GstRtspServer.RTSPServer):
    def __init__(self):
        super(GstServer, self).__init__()
        factory = SharedSensorFactory()
        self.get_mount_points().add_factory("/test", factory)
        self.attach(None)

server = GstServer()
loop = GObject.MainLoop()
loop.run()

关键修改说明 #

1. 公共流水线与分流设计：将视频采集、格式转换、H264编码作为公共部分，仅启动一次，通过tee元素将编码后的流分发给所有客户端
2. 客户端专属分支：每个客户端仅创建queue（缓解流控压力）和rtph264pay（RTSP协议打包），避免重复编码
3. 资源自动释放：监听客户端断开事件，动态释放tee的请求pad，防止资源泄漏
4. 移除OpenCV依赖：直接使用GStreamer原生元素构建流水线，减少中间环节的性能开销

内容的提问来源于stack exchange，提问作者张金阳