GStreamer appsink对接RTSP服务appsrc推流出现2秒以上高延迟问题

算力阿峰

2026-8-30

GStreamer RTSP推流appsrc缓存导致超2秒延迟问题

问题现象

搭建的GStreamer推流链路逻辑：上游采集编码管线通过appsink拉取编码后的H264视频样本，将样本推送至RTSP服务媒体管线的源元素appsrc，基础功能运行正常，客户端连接RTSP地址可正常观看推流视频，但存在超过2秒的严重播放延迟。
排查观测结论：appsrc内部队列缓存大量缓冲区；上游管线启动后，appsrc的源pad不会立即开始读取数据，从管线启动到appsrc开始向外读取数据的时间差，直接转化为最终播放端的固定延迟。
观测验证依据：每次向appsrc推送缓冲区时，都会读取appsrc内部队列当前缓存的字节数，管线启动后的一段时间内该数值持续上升；待appsrc开始向外读取数据后，队列缓存字节数在后续推流过程中始终维持稳定值，不再持续增长。

测试验证代码

#include <stdio.h>
#include <gst/gst.h>
#include <gst/app/gstappsrc.h>
#include <gst/app/gstappsink.h>
#include <time.h>

#include <gst/rtsp-server/rtsp-server.h>

  GMainLoop *loop;
  GstElement *appsink;
  GstElement *appsrc;
  GstElement *appsink_pipeline;

/* 以下函数用于以可读格式打印Capabilities */
static gboolean print_field (GQuark field, const GValue * value, gpointer pfx) {
  gchar *str = gst_value_serialize (value);

  g_print ("%s  %15s: %s\n", (gchar *) pfx, g_quark_to_string (field), str);
  g_free (str);
  return TRUE;
}

static void print_caps (const GstCaps * caps, const gchar * pfx) {
  guint i;

  g_return_if_fail (caps != NULL);

  if (gst_caps_is_any (caps)) {
    g_print ("%sANY\n", pfx);
    return;
  }
  if (gst_caps_is_empty (caps)) {
    g_print ("%sEMPTY\n", pfx);
    return;
  }

  for (i = 0; i < gst_caps_get_size (caps); i++) {
    GstStructure *structure = gst_caps_get_structure (caps, i);

    g_print ("%s%s\n", pfx, gst_structure_get_name (structure));
    gst_structure_foreach (structure, print_field, (gpointer) pfx);
  }
}

/* appsink通知有新缓冲区可处理时触发的回调 */
static GstFlowReturn
on_new_sample_from_sink (GstElement * elt, void * data)
{
  GstSample *sample;
  GstFlowReturn ret = GST_FLOW_OK;
  guint64 bytes;
 

  /* 从appsink拉取样本 */
  sample = gst_app_sink_pull_sample (GST_APP_SINK (elt));
  if(appsrc)
  {

    bytes = gst_app_src_get_current_level_bytes(GST_APP_SRC(appsrc));
    g_print("buffered bytes before push %lu\n", bytes);
    
    ret = gst_app_src_push_sample(GST_APP_SRC (appsrc), sample);
  }
  
  gst_sample_unref (sample);
 
  return ret;
}
 
/* 上游源管线收到GstMessage（尤其是EOS）时触发的回调，同步通知appsrc */
static gboolean
on_source_message (GstBus * bus, GstMessage * message, void * data)
{
  gint percent;
  g_print ("%s\n", __func__);
 
  switch (GST_MESSAGE_TYPE (message)) {
    case GST_MESSAGE_EOS:
      g_print ("The source got dry\n");
      gst_app_src_end_of_stream (GST_APP_SRC (appsrc));
      break;
    case GST_MESSAGE_ERROR:
      g_print ("Received error\n");
      g_main_loop_quit (loop);
      break;
    case GST_MESSAGE_BUFFERING:
      gst_message_parse_buffering (message, &percent);
      g_print ("Buffering = %d\n", percent);
      break;
    default:
      break;
  }
  return TRUE;
}
 
static GstFlowReturn need_data (GstElement * appsrc_loc,
                    guint length,
                    gpointer udata)
{

  g_print("Need data\n");

  return GST_FLOW_OK;

}

/* 定时清理会话池中过期会话，当前需显式调用，后续版本可能合并到主循环自动执行 */
static gboolean
timeout (GstRTSPServer * server)
{
  GstRTSPSessionPool *pool;

  pool = gst_rtsp_server_get_session_pool (server);
  gst_rtsp_session_pool_cleanup (pool);
  g_object_unref (pool);

  return TRUE;
}

void clientConnected(GstRTSPServer* server, GstRTSPClient* client, gpointer user)
{
  g_print("%s\n", __func__);
  
}

static void media_state_cb(GstRTSPMedia *media, GstState state)
{

  g_print("media state = %d\n", state);

}


static void
media_construct (GstRTSPMediaFactory * factory, GstRTSPMedia * media,
    gpointer user_data)
{
  GstElement *element;

  g_print("%s\n", __func__);
  /* 获取提供媒体流的元素 */
  element = gst_rtsp_media_get_element (media);

  /* 获取命名为appsrc的源元素 */
  appsrc = gst_bin_get_by_name_recurse_up (GST_BIN (element), "appsrc");
  g_signal_connect (appsrc, "need-data",
      G_CALLBACK (need_data), NULL);

  g_signal_connect (media, "new-state",
      G_CALLBACK (media_state_cb), NULL);

  gst_object_unref (element);
}

static void
media_configure (GstRTSPMediaFactory * factory, GstRTSPMedia * media,
    gpointer user_data)
{
  GstPad *pad;
  GstCaps *caps;
  gchar *caps_str;
  GstElement *element;

  g_print("%s\n", __func__);

  /* 获取提供媒体流的元素 */
  element = gst_rtsp_media_get_element (media);

  /* 获取命名为appsrc的源元素 */
  appsrc = gst_bin_get_by_name_recurse_up (GST_BIN (element), "appsrc");

  pad = gst_element_get_static_pad (appsink, "sink");
  if(pad)
  {
    g_print("Got pad\n");
    caps = gst_pad_get_current_caps (pad);
    if(caps)
    {
      caps_str = gst_caps_to_string  (caps);
      g_print("Got caps %s\n", caps_str);
      g_object_set (G_OBJECT (appsrc), "caps",  caps, NULL);

      gst_caps_unref(caps);
    }
  }

  /* 配置appsrc使用时间格式处理缓冲区 */
  gst_util_set_object_arg (G_OBJECT (appsrc), "format", "time");

  gst_object_unref (element);
}

int main (int argc, char *argv[]){
  GstBus *bus;
  GstRTSPServer *server;
  GstRTSPMountPoints *mounts;
  GstRTSPMediaFactory *factory;

  gchar src[] = "nvv4l2camerasrc device=/dev/video0  ! video/x-raw(memory:NVMM), width=1920, height=1080, format=UYVY, framerate=60/1 ! " 
        " queue max-size-buffers=3 leaky=downstream ! "
        " nvvidconv name=conv ! video/x-raw(memory:NVMM), width=1280, height=720, format=NV12, framerate=60/1 ! "
        " nvv4l2h264enc control-rate=1  bitrate=8000000 preset-level=1 profile=0 disable-cabac=1 maxperf-enable=1 name=encoder insert-sps-pps=1 insert-vui=1  idrinterval=30 ! "
        " appsink name=appsink sync=false max-buffers=3";


  gchar sink[] = "( appsrc name=appsrc format=3 stream-type=0 is-live=true blocksize=2097152  max-bytes=200000 ! "
                " queue max-size-buffers=3 leaky=no ! "
                " rtph264pay config-interval=1 name=pay0 )";
                  
        
  gst_init (&argc, &argv);

  loop = g_main_loop_new (NULL, FALSE);

  /* 创建带appsink的采集管线 */
  g_print("Creating pipeline with appsink\n");
  appsink_pipeline = gst_parse_launch (src, NULL);
 
  if (appsink_pipeline == NULL) {
    g_print ("Bad source\n");
    g_main_loop_unref (loop);
    return -1;
  }

  /* 监听管线总线消息，主要处理EOS */
  bus = gst_element_get_bus (appsink_pipeline);
  gst_bus_add_watch (bus, (GstBusFunc) on_source_message, appsink_pipeline);
  gst_object_unref (bus);

  /* 为appsink注册新样本回调 */
  g_print("Creating push buffer callback\n");
  appsink = gst_bin_get_by_name (GST_BIN (appsink_pipeline), "appsink");
  g_object_set (G_OBJECT (appsink), "emit-signals", TRUE, "sync", FALSE, NULL);
  g_signal_connect (appsink, "new-sample",
      G_CALLBACK (on_new_sample_from_sink), NULL);

  /* 创建以appsrc为源的RTSP服务 */
  g_print("Creating rtsp server\n");
  /* 初始化RTSP服务实例 */
  server = gst_rtsp_server_new ();

  /* 获取服务挂载点，用于URI和媒体工厂的映射 */
  mounts = gst_rtsp_server_get_mount_points (server);

  /* 创建媒体工厂，通过gst-launch语法构建管线，管线中需包含命名为pay%d的RTP打包元素 */
  factory = gst_rtsp_media_factory_new ();

  gst_rtsp_media_factory_set_launch (factory, sink);
  gst_rtsp_media_factory_set_shared(factory, TRUE);

  /* 将媒体工厂挂载到/test路径 */
  gst_rtsp_mount_points_add_factory (mounts, "/test", factory);

  /* 释放挂载点对象引用 */
  g_object_unref (mounts);

  /* 将服务挂载到默认主上下文 */
  if (gst_rtsp_server_attach (server, NULL) == 0)
    goto failed;

  /* 注册定时任务清理过期会话 */
  g_timeout_add_seconds (2, (GSourceFunc) timeout, server);

  g_signal_connect (server, "client-connected",
      G_CALLBACK (clientConnected), NULL);

  /* 注册media-constructed回调获取appsrc实例 */
  g_print("Creating media-constructed callback\n");

  g_signal_connect (factory, "media-constructed", (GCallback) media_construct,
      NULL);

  g_signal_connect (factory, "media-configure", (GCallback) media_configure,
      NULL);

  /* 启动主循环 */ 

  g_print("Running main loop\n");
  
  gst_element_set_state (appsink_pipeline, GST_STATE_PLAYING);
  g_main_loop_run (loop);
  gst_element_set_state (appsink_pipeline, GST_STATE_NULL);

  return 0;

  /* 错误处理 */
failed:
  {
    g_print ("failed to attach the server\n");
    return -1;
  }
}

问题诱因

启动时序逻辑错误：当前代码在RTSP媒体管线未完成初始化、appsrc未进入数据拉取状态时，就提前将上游采集编码管线切换到PLAYING状态开始推送编码帧。此时appsrc的源pad未启动数据消费逻辑，所有推送过来的帧全部缓存在内部队列中，直到第一个客户端连接、媒体管线正式启动后才开始读取队列数据。从管线启动到appsrc开始读数据的时间差有多长，队列里就攒了多长时间的历史帧，这部分帧会被按顺序推到客户端，直接形成固定的初始延迟，和观测到的现象完全吻合。
参数配置进一步放大延迟：当前appsrc配置的max-bytes=200000（约200KB），在8Mbps的编码码率下可缓存约200ms数据，若启动后等待客户端连接的时间达到2秒，队列会持续缓存直到达到阈值后阻塞推流，最终形成2秒左右的固定延迟；同时appsrc后方的队列配置为leaky=no，不丢弃任何旧帧，一旦出现消费速度和生产速度的微小波动，队列就会持续攒数据拉高延迟。
服务端默认延迟配置未调整：GStreamer RTSP服务默认会预留一定的RTP缓存时间，客户端jitter buffer也会默认配置百毫秒级的缓存，这部分会和前面的初始延迟叠加，但不是导致启动阶段队列缓存持续上涨的核心原因。

可行解决方案

修正启动时序
- 取消main函数里提前启动上游采集管线的逻辑，改为在media-configure回调触发、确认appsrc完成caps配置、媒体管线即将启动时，再将上游采集管线切到PLAYING状态。由于已经配置了媒体共享（set_shared=TRUE），仅需在第一个媒体实例创建完成后启动一次上游管线即可，从根源上避免提前推数据攒缓存的问题。
- 给appsrc绑定enough-data和need-data信号：队列缓存达到阈值触发enough-data时，临时阻塞上游appsink的帧拉取；队列缓存不足触发need-data时再恢复拉取，避免队列无限制堆积。
调整低延迟相关参数
- 优化appsrc配置：将max-bytes下调到50000_100000（50100KB），设置min-percent=0让appsrc拿到数据就立即推送，无需等缓存攒到指定比例；关闭do-timestamp，直接复用上游编码输出的原始时间戳，避免appsrc重新打戳引入的时间偏移。
- 修改管线队列配置：将appsrc后方的queue元素改为leaky=downstream模式，设置max-size-buffers=1，队列满时自动丢弃最旧的缓存帧，仅保留最新帧，避免队列积压。
- 调整RTSP服务延迟参数：在media-configure回调中调用gst_rtsp_media_set_latency(media, 0)，将服务端RTP传输缓存延迟降到最低。
主动清理无效缓存
- 若业务要求必须提前启动上游采集管线，在media-configure回调拿到appsrc实例后，先调用gst_app_src_flush()清空appsrc内部已经堆积的所有历史缓存，再开始正常推流，直接消除启动阶段攒下的延迟。
- 推流逻辑增加主动丢帧判断：每次推送样本前先查询appsrc当前缓存字节数，若缓存大小超过单帧平均大小，直接丢弃当前待推送的旧帧，始终保证队列内缓存的帧不超过1个，最终端到端延迟可控制在1_{2帧（60fps下约17}34ms）。

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

GStreamer appsink对接RTSP服务appsrc推流出现2秒以上高延迟问题

问题现象 #

测试验证代码 #

问题诱因 #

可行解决方案 #

问题现象

测试验证代码

问题诱因

可行解决方案