如何使用GStreamer输出支持多客户端访问的RTSP流及报错解决

你遇到的Assertion fctx->async_lock failed at libavcodec/pthread_frame.c:155错误属于多线程并发访问编码器的线程安全冲突：

• 你开启了set_shared(True)配置，多个客户端会共享同一个媒体工厂实例，但每个客户端连接会触发独立的编码流水线运行
• 多个客户端的need-data信号会并发调用on_need_data方法，同时读取共享的VideoCapture实例、同时调用x264编码器，触发libavcodec内部的锁断言失败

按以下步骤修改代码即可解决：

1. 给x264编码器添加单线程运行参数，避免编码器内部多线程冲突
2. 给帧读取、推送逻辑加全局线程锁，避免多线程同时操作帧资源
3. 可选：如果客户端数量极少，也可以直接关闭工厂共享，每个客户端单独运行全量流水线

import cv2
import gi
import threading

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

class SensorFactory(GstRtspServer.RTSPMediaFactory):
    def __init__(self, **properties):
        super(SensorFactory, self).__init__(**properties)
        self.cap = cv2.VideoCapture("http://192.168.2.153/video/mjpg.cgi")
        self.number_frames = 0
        self.fps = 10
        self.lock = threading.Lock() # 新增线程锁
        self.duration = 1 / self.fps * Gst.SECOND  # duration of a frame in nanoseconds
        # x264enc新增threads=1参数
        self.launch_string = 'appsrc name=source is-live=true block=true format=GST_FORMAT_TIME ' \
                             'caps=video/x-raw,format=BGR,width=640,height=480,framerate={}/1 ' \
                             '! videoconvert ! video/x-raw,format=I420 ' \
                             '! x264enc speed-preset=ultrafast tune=zerolatency threads=1 ' \
                             '! rtph264pay config-interval=1 name=pay0 pt=96'.format(self.fps)

    def on_need_data(self, src, lenght):
        # 加锁保证同一时间只有一个线程操作读帧、推帧逻辑
        with self.lock:
            if self.cap.isOpened():
                ret, frame = self.cap.read()
                if ret:
                    data = frame.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
                    retval = src.emit('push-buffer', buf)

                    print ('pushed buffer, frame {}, duration {} ns, durations {} s'.format(self.number_frames, self.duration, self.duration / Gst.SECOND))

                    if retval != Gst.FlowReturn.OK:
                        print(retval)

    def do_create_element(self, url):
        return Gst.parse_launch(self.launch_string)

    def do_configure(self, rtsp_media):
        self.number_frames = 0
        appsrc = rtsp_media.get_element().get_child_by_name('source')
        appsrc.connect('need-data', self.on_need_data)


class GstServer(GstRtspServer.RTSPServer):
    def __init__(self, **properties):
        super(GstServer, self).__init__(**properties)
        self.factory = SensorFactory()
        self.factory.set_shared(True)
        self.get_mount_points().add_factory("/test", self.factory)
        self.set_service('9999')
        self.attach(None)

Gst.init(None)
server = GstServer()
loop = GLib.MainLoop()
loop.run()

内容的提问来源于stack exchange，提问作者scy-ai