参考：

https://zhuanlan.zhihu.com/p/370467867

https://github.com/mad4ms/python-opencv-gstreamer-examples/blob/master/gst_rtsp_server.py

主要提示：

1、服务端必须要用sudo 权限，否则无法启动端口！！！

2、output.Open()打开推流端口 用if output.IsStreaming(): 判断是否成功。需要服务完全启动，否则会失败！

3、服务端必须要推送渲染图像，否则仅仅打开服务，接收者流。前台播放器是无法播放的。

4、输出分辨率需要为480，640，720，1080这些标准分辨率，不能是非常正常分辨率

5、输出的帧率，只受输出分辨率影响，不受输入视频分辨率和数量影响

6、在推送output.Render(cudaFrame)推送后必须要增加time.sleep(0.03)延迟，否则会导致前端无法播放，如果前端无法播放，则print("xx")输出下任意字符，就好了。

服务端

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

from func_timeout import func_set_timeout
import jetson.utils
import jetson.inference
import numpy as np
import threading
import time
import random


#非常重要
GObject.threads_init()
Gst.init(None)

class SimpleRtspServer:
    def __init__(self,port=8554,path="/1"):
        self.server = GstRtspServer.RTSPServer.new()
        self.server.set_service(str(port))      
        self.server.attach(None)

        self.factory = GstRtspServer.RTSPMediaFactory.new()
        self.factory.set_launch('(udpsrc name=pay0 port=5401 buffer-size=524288 \
                                   caps="application/x-rtp, media=video, clock-rate=90000, \
                                   encoding-name=(string)H264, payload=96 ")')
        self.factory.set_shared(True)
        
        self.server.get_mount_points().add_factory(path, self.factory)

        self.thread = None
        self.running = False

    def start(self):
        self.thread = threading.Thread(target=self._run)
        self.thread.start()

    def _run(self):
        output = jetson.utils.videoOutput('rtp://@:5401',argv=["--headless", "--bitrate=4096000"])

        while True:
            if not output.Open():
                time.sleep(1)

            if output.IsStreaming():
                break

        frame = np.zeros((720,1280, 3), dtype=np.uint8)

        tc=time.time()
        while True:
            #随机颜色
            frame[100:200, 100:200] = [random.randint(0,255),random.randint(0,255),random.randint(0,255)]
            img_cuda = jetson.utils.cudaFromNumpy(frame)
  
            self.render(output,img_cuda)
            tc = (1 / 25) - (time.time() - tc)
            if tc > 0:
                time.sleep(tc)

    @func_set_timeout(1/25)
    def render(self, output, cudaFrame):
        output.Render(cudaFrame)


if __name__ == "__main__":
    rtsp_server = SimpleRtspServer()
    rtsp_server.start()

    #非常重要
    loop = GObject.MainLoop()
    loop.run()

客户端

import cv2
import time

def test_rtsp_with_opencv(rtsp_url, timeout=30):
    """
    使用OpenCV测试RTSP流连接
    """
    print(f"正在测试RTSP流: {rtsp_url}")
    
    cap = cv2.VideoCapture(rtsp_url)
    
    if not cap.isOpened():
        print("❌ 无法打开RTSP流")
        return False
    
    print("✅ RTSP流打开成功")
    
    # 设置超时时间
    start_time = time.time()
    frames_received = 0
    
    while time.time() - start_time < timeout:
        ret, frame = cap.read()
        
        if ret:
            frames_received += 1
            print(f"✅ 接收到第 {frames_received} 帧 - 分辨率: {frame.shape[1]}x{frame.shape[0]}")
            
            # 显示第一帧信息
            if frames_received == 1:
                cv2.imwrite('test_frame.jpg', frame)
                print("✅ 第一帧已保存为 test_frame.jpg")
            
            # 每秒显示一次进度
            if frames_received % 30 == 0:
                print(f"⏰ 已接收 {frames_received} 帧")
                
        else:
            print("❌ 读取帧失败")
            break
        
        # 按'q'退出测试
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
    
    cap.release()
    cv2.destroyAllWindows()
    
    print(f"\n测试结果:")
    print(f"总接收帧数: {frames_received}")
    print(f"平均帧率: {frames_received/timeout:.2f} fps")
    
    if frames_received > 0:
        print("✅ RTSP流测试成功")
        return True
    else:
        print("❌ RTSP流测试失败")
        return False

if __name__ == "__main__":
    # 测试本地RTSP服务器
    rtsp_url = "rtsp://localhost:8554/1"
    test_rtsp_with_opencv(rtsp_url)

成功