如何用GStreamer与OpenCV搭建支持多客户端的RTSP组播服务器？

你的猜测是正确的，当前代码存在几个关键问题，导致后续客户端无法正常接收流：

核心问题 #

• 共享模式下重复创建流水线：虽然设置了set_shared(True)，但SensorFactory会为每个客户端创建独立的GStreamer编码推送流水线，不仅浪费资源，还会导致组播流冲突。
• 时间戳全局重置：number_frames是工厂实例变量，每次新客户端连接时do_configure会重置它，导致后续客户端的流时间戳不连续，播放器无法正常解码。
• 组播地址范围不规范：使用的224.0.0.1到240.0.0.10包含非通用组播段，应该使用IANA分配的合法本地组播地址（如224.0.1.0~224.0.1.255）。
• 单帧多实例推送：每个客户端的appsrc都会触发on_need_data，导致同一帧被多次编码推送，而非共享编码后的流。

改进后的代码 #

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

# 假设frame_list是全局共享的线程安全列表
frame_list = Manager.list()
# 替换为你的实际帧源
frame_list.append(cv2.imread("test_frame.jpg"))

class SharedSensorFactory(GstRtspServer.RTSPMediaFactory):
    def __init__(self, image_type):
        super(SharedSensorFactory, self).__init__()
        self.image_type = image_type
        self.fps = 30.0
        self.duration = 1.0 / self.fps * Gst.SECOND
        self.number_frames = 0
        
        # 创建全局唯一的编码+组播推送流水线
        self.pipeline = Gst.parse_launch(
            'appsrc name=source is-live=true format=GST_FORMAT_TIME '
            'caps=video/x-raw,format=BGR,width=640,height=360,framerate=30/1 '
            '! videoconvert ! video/x-raw,format=I420 '
            '! x264enc speed-preset=ultrafast tune=zerolatency '
            '! rtph264pay name=pay0 pt=96 '
            '! udpsink host=224.0.1.1 port=5004 auto-multicast=true ttl=5'
        )
        self.appsrc = self.pipeline.get_child_by_name('source')
        self.appsrc.connect('need-data', self.on_need_data)
        self.pipeline.set_state(Gst.State.PLAYING)

    def do_create_element(self, url):
        # 客户端连接时，创建从组播地址拉流的流水线，共享同一编码源
        return Gst.parse_launch(
            'udpsrc address=224.0.1.1 port=5004 '
            '! application/x-rtp,media=video,encoding-name=H264,payload=96 '
            '! rtph264depay ! h264parse ! decodebin ! videoconvert ! autovideosink'
        )

    def on_need_data(self, src, length):
        # 仅通过全局流水线推送帧，所有客户端共享该流
        if len(frame_list) <= self.image_type:
            return
        data = cv2.resize(frame_list[self.image_type], (640, 360), interpolation=cv2.INTER_LINEAR)
        data = data.tobytes()
        buf = Gst.Buffer.new_allocate(None, len(data), None)
        buf.fill(0, data)
        buf.duration = self.duration
        timestamp = self.number_frames * self.duration
        buf.pts = buf.dts = int(timestamp)
        buf.offset = timestamp
        self.number_frames += 1
        src.emit('push-buffer', buf)

    def do_configure(self, rtsp_media):
        # 无需重置时间戳，全局流水线持续维护连续时间戳
        pass

class MulticastServer(GstRtspServer.RTSPServer):
    def __init__(self):
        GstRtspServer.RTSPServer.__init__(self)
        self.set_service("5000")

        self.factory0 = SharedSensorFactory(0)
        self.factory0.set_shared(True)
        self.factory0.set_eos_shutdown(True)

        # 配置合法的组播地址池
        self.address_pool = GstRtspServer.RTSPAddressPool.new()
        self.address_pool.add_range("224.0.1.1", "224.0.1.10", 5004, 5014, 5)
        self.factory0.set_address_pool(self.address_pool)
        self.get_mount_points().add_factory("/stream", self.factory0)

Gst.init(None)
server = MulticastServer()
server.attach(None)

GLib.MainLoop().run()

关键改进说明 #

• 单编码流水线：只创建一个编码和组播推送的流水线，所有客户端通过RTSP获取组播地址后，直接从组播地址拉流，避免重复编码和资源浪费。
• 连续时间戳维护：number_frames不再被重置，保证流的时间戳连续，播放器能正常解码。
• 规范组播配置：使用224.0.1.x段的合法本地组播地址，同时设置合理的TTL值（当前为5，可根据网络规模调整）。
• 客户端流水线简化：客户端仅负责拉流和解码，共享同一个组播源。

额外注意事项 #

• 确保服务器与客户端在同一局域网内，组播流量可正常传输。
• 保证frame_list的写入操作是线程安全的，避免帧数据损坏。

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