如何从同步RTP流中获取GStreamer匹配缓冲区并确保数据同步？

核心实现思路 #

通过rtspsrc拉取同一服务器的两路流，分别走独立处理分支：一路解码H.264得到原始帧，另一路直接解析未压缩原始流，依赖RTSP服务器给对应帧打上的**相同PTS（显示时间戳）**做匹配，最后用appsink提取缓冲区数据。

管道配置要点 #

• rtspsrc需启用多路流处理，设置latency参数控制延迟，保障同步性
• H.264流处理分支：rtspsrc ! rtph264depay ! h264parse ! avdec_h264 ! videoconvert ! appsink name=h264_sink
• 未压缩原始流处理分支（以YUYV格式为例）：rtspsrc ! rtpgenericdepay ! videoparse format=yuyv width=XXX height=XXX framerate=XX/1 ! appsink name=raw_sink
• 无需重复创建rtspsrc实例，同一组件会自动处理服务器推送的多路流

同步匹配缓冲区的关键逻辑 #

两路流来自同一摄像头，服务器会给同一时刻的帧分配相同PTS，以此为匹配核心：

• 在appsink的回调函数中，通过gst_buffer_get_pts()获取每个缓冲区的时间戳
• 维护两个缓存队列，分别存储H.264解码帧和原始流帧的PTS与缓冲区
• 当某一路的帧到来时，在另一路的缓存中查找PTS匹配的帧；若缓存中无匹配帧则暂存，同时清理过期的旧帧（PTS远小于当前帧的）

C语言代码实现示例 #

#include <gst/gst.h>
#include <glib.h>

// 缓存结构体：存储帧缓冲区与对应PTS
typedef struct {
    GstBuffer *buf;
    GstClockTime pts;
} FrameCache;

GQueue *h264_frame_cache = NULL;
GQueue *raw_frame_cache = NULL;

// H.264解码流的appsink回调
static GstFlowReturn h264_sink_new_sample(GstAppSink *sink, gpointer user_data) {
    GstBuffer *buf = gst_app_sink_pull_buffer(sink);
    if (!buf) return GST_FLOW_ERROR;

    GstClockTime pts = gst_buffer_get_pts(buf);
    FrameCache *cache_item = g_malloc(sizeof(FrameCache));
    cache_item->buf = gst_buffer_ref(buf);
    cache_item->pts = pts;
    g_queue_push_tail(h264_frame_cache, cache_item);

    // 尝试匹配原始流的对应帧
    GList *iter = g_queue_peek_head_link(raw_frame_cache);
    while (iter) {
        FrameCache *raw_item = iter->data;
        if (raw_item->pts == pts) {
            // 找到匹配帧，提取数据
            g_print("匹配到同步帧，PTS: %" GST_TIME_FORMAT "\n", GST_TIME_ARGS(pts));
            
            // 读取H.264解码后的原始数据
            GstMapInfo h264_map;
            if (gst_buffer_map(cache_item->buf, &h264_map, GST_MAP_READ)) {
                // h264_map.data即为解码后的原始视频数据，可在此处处理
                gst_buffer_unmap(cache_item->buf, &h264_map);
            }

            // 读取未压缩原始流数据
            GstMapInfo raw_map;
            if (gst_buffer_map(raw_item->buf, &raw_map, GST_MAP_READ)) {
                // raw_map.data即为未压缩原始视频数据，可在此处处理
                gst_buffer_unmap(raw_item->buf, &raw_map);
            }

            // 清理缓存
            g_queue_delete_link(raw_frame_cache, iter);
            g_queue_pop_head(h264_frame_cache);
            gst_buffer_unref(cache_item->buf);
            gst_buffer_unref(raw_item->buf);
            g_free(cache_item);
            g_free(raw_item);
            break;
        } else if (raw_item->pts < pts) {
            // 清理过期帧
            iter = g_queue_next(iter);
            g_queue_pop_head(raw_frame_cache);
            gst_buffer_unref(raw_item->buf);
            g_free(raw_item);
        } else {
            // 原始流帧未到，退出等待
            break;
        }
    }

    gst_buffer_unref(buf);
    return GST_FLOW_OK;
}

// 原始流的appsink回调，逻辑与H.264分支反向匹配
static GstFlowReturn raw_sink_new_sample(GstAppSink *sink, gpointer user_data) {
    GstBuffer *buf = gst_app_sink_pull_buffer(sink);
    if (!buf) return GST_FLOW_ERROR;

    GstClockTime pts = gst_buffer_get_pts(buf);
    FrameCache *cache_item = g_malloc(sizeof(FrameCache));
    cache_item->buf = gst_buffer_ref(buf);
    cache_item->pts = pts;
    g_queue_push_tail(raw_frame_cache, cache_item);

    // 尝试匹配H.264解码帧，逻辑同上方回调，反向查找即可

    gst_buffer_unref(buf);
    return GST_FLOW_OK;
}

// 处理rtspsrc动态生成的pad连接
static void on_rtspsrc_pad_added(GstElement *src, GstPad *new_pad, gpointer pipeline) {
    GstElement *h264_depay = gst_bin_get_by_name(GST_BIN(pipeline), "h264-depay");
    GstElement *raw_depay = gst_bin_get_by_name(GST_BIN(pipeline), "raw-depay");

    GstPad *h264_sink_pad = gst_element_get_static_pad(h264_depay, "sink");
    GstPad *raw_sink_pad = gst_element_get_static_pad(raw_depay, "sink");

    GstCaps *caps = gst_pad_get_current_caps(new_pad);
    const gchar *caps_str = gst_caps_to_string(caps);

    if (g_strstr_len(caps_str, -1, "H264")) {
        if (!gst_pad_is_linked(h264_sink_pad)) {
            GstPadLinkReturn ret = gst_pad_link(new_pad, h264_sink_pad);
            if (GST_PAD_LINK_FAILED(ret)) {
                g_printerr("H.264流Pad连接失败\n");
            }
        }
    } else if (g_strstr_len(caps_str, -1, "rawvideo")) {
        if (!gst_pad_is_linked(raw_sink_pad)) {
            GstPadLinkReturn ret = gst_pad_link(new_pad, raw_sink_pad);
            if (GST_PAD_LINK_FAILED(ret)) {
                g_printerr("原始流Pad连接失败\n");
            }
        }
    }

    // 资源清理
    gst_caps_unref(caps);
    gst_object_unref(h264_sink_pad);
    gst_object_unref(raw_sink_pad);
    gst_object_unref(h264_depay);
    gst_object_unref(raw_depay);
}

int main(int argc, char *argv[]) {
    gst_init(&argc, &argv);

    h264_frame_cache = g_queue_new();
    raw_frame_cache = g_queue_new();

    // 创建GStreamer元素
    GstElement *pipeline = gst_pipeline_new("dual-rtsp-stream-pipeline");
    GstElement *rtspsrc = gst_element_factory_make("rtspsrc", "rtsp-source");
    GstElement *rtph264depay = gst_element_factory_make("rtph264depay", "h264-depay");
    GstElement *h264parse = gst_element_factory_make("h264parse", "h264-parse");
    GstElement *avdec_h264 = gst_element_factory_make("avdec_h264", "h264-decoder");
    GstElement *videoconvert = gst_element_factory_make("videoconvert", "video-convert");
    GstElement *h264_appsink = gst_element_factory_make("appsink", "h264-appsink");
    GstElement *rtpgenericdepay = gst_element_factory_make("rtpgenericdepay", "raw-depay");
    GstElement *videoparse = gst_element_factory_make("videoparse", "raw-parse");
    GstElement *raw_appsink = gst_element_factory_make("appsink", "raw-appsink");

    if (!pipeline || !rtspsrc || !rtph264depay || !h264parse || !avdec_h264 || !videoconvert || !h264_appsink || !rtpgenericdepay || !videoparse || !raw_appsink) {
        g_printerr("无法创建全部GStreamer元素\n");
        return -1;
    }

    // 配置rtspsrc：替换为你的RTSP服务器地址，调整延迟
    g_object_set(G_OBJECT(rtspsrc), "location", "rtsp://your-server-ip/stream-path", "latency", 200, NULL);
    // 配置videoparse：根据服务器输出的原始流格式调整参数
    g_object_set(G_OBJECT(videoparse), "format", GST_VIDEO_FORMAT_YUY2, "width", 1920, "height", 1080, "framerate", GST_FRACTION(30,1), NULL);

    // 配置appsink启用回调
    g_object_set(G_OBJECT(h264_appsink), "emit-signals", TRUE, "sync", FALSE, NULL);
    g_signal_connect(h264_appsink, "new-sample", G_CALLBACK(h264_sink_new_sample), NULL);
    g_object_set(G_OBJECT(raw_appsink), "emit-signals", TRUE, "sync", FALSE, NULL);
    g_signal_connect(raw_appsink, "new-sample", G_CALLBACK(raw_sink_new_sample), NULL);

    // 添加元素到管道
    gst_bin_add_many(GST_BIN(pipeline), rtspsrc, rtph264depay, h264parse, avdec_h264, videoconvert, h264_appsink, rtpgenericdepay, videoparse, raw_appsink, NULL);

    // 静态链接分支元素
    gst_element_link_many(rtph264depay, h264parse, avdec_h264, videoconvert, h264_appsink, NULL);
    gst_element_link_many(rtpgenericdepay, videoparse, raw_appsink, NULL);

    // 监听rtspsrc的pad添加信号，处理动态流连接
    g_signal_connect(rtspsrc, "pad-added", G_CALLBACK(on_rtspsrc_pad_added), pipeline);

    // 启动管道
    gst_element_set_state(pipeline, GST_STATE_PLAYING);

    // 运行主循环
    GMainLoop *main_loop = g_main_loop_new(NULL, FALSE);
    g_main_loop_run(main_loop);

    // 资源清理
    gst_element_set_state(pipeline, GST_STATE_NULL);
    gst_object_unref(GST_OBJECT(pipeline));
    g_main_loop_unref(main_loop);
    g_queue_free_full(h264_frame_cache, (GDestroyNotify)g_free);
    g_queue_free_full(raw_frame_cache, (GDestroyNotify)g_free);

    return 0;
}

关键注意事项 #

• 必须根据RTSP服务器输出的原始流格式，调整videoparse的format、width、height、framerate参数，否则会导致流解析失败
• latency参数需根据网络情况调整：太小易丢帧，太大则延迟过高
• 若两路流的PTS存在微小偏差（如±10ms），可修改匹配逻辑，设置容忍阈值而非严格相等
• 缓存队列的长度需合理控制，避免内存占用过高，同时能应对网络抖动导致的帧顺序错乱

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