GStreamer删除Queue分支致其他分支暂停，Appsrc偶发拒接数据排查

安全阿岳

2026-8-1

问题描述

通过动态连接/断开Tee分支的方式控制视频录制，小概率出现appsrc拒绝接收数据、无法继续工作的情况，无法稳定复现。Pipeline结构为：appsrc → ... → tee（一路用于预览，一路动态创建用于录制：queue → h264parse → qtmux → filesink），控制录制的核心代码如下：

void GstHandle::startRecord(QString path)
{
    if(mIsRecording){
        log_info("%s : GST current record is already open", __func__);
        return;
    }
    record_start = QDateTime::currentMSecsSinceEpoch();
    mRecordQueue = gst_element_factory_make("queue", "record_queue");
    mH264Parse = gst_element_factory_make("h264parse", "myparse");
    mQtMux = gst_element_factory_make("qtmux", "qtmux");
    mFileSink = gst_element_factory_make("filesink", "filesink");

    if(!mFileSink || !mRecordQueue || !mQtMux || !mH264Parse){
        log_error("%s : GST create elements error", __func__);
        return;
    }

    g_object_set (mFileSink, "location", path.toLocal8Bit().data(), NULL);

    gst_bin_add_many(GST_BIN(mPipeline), mRecordQueue, mH264Parse, mQtMux, mFileSink, NULL);
    gst_element_sync_state_with_parent(mFileSink);
    gst_element_sync_state_with_parent(mQtMux);
    gst_element_sync_state_with_parent(mH264Parse);
    gst_element_sync_state_with_parent(mRecordQueue);
    if(gst_element_link_many(mRecordQueue, mH264Parse, mQtMux, mFileSink, NULL) != TRUE)
    {
        log_error("%s : GST video queue link error", __func__);
    }

    gst_element_sync_state_with_parent(mRecordQueue);

    //set pads
    tee_record_pad = gst_element_get_request_pad(mTee, "src_2");
    record_sink_pad = gst_element_get_static_pad(mRecordQueue, "sink");
    if(!tee_record_pad || !record_sink_pad){
        log_error("%s : GST get pads error", __func__);
        return;
    }
    GstPadLinkReturn ret = GST_PAD_LINK_OK;
    if(!gst_pad_is_linked(record_sink_pad)){
        ret = gst_pad_link(tee_record_pad, record_sink_pad);
        if(ret != GST_PAD_LINK_OK)
        {
            log_error("%s : GST pad link error", __func__);
            gst_object_unref(record_sink_pad);
            return;
        }
    }
    gst_object_unref(record_sink_pad);

    if(gst_element_set_state(mPipeline, GST_STATE_PLAYING) == GST_STATE_CHANGE_FAILURE)
    {
        log_error("%s : GST Unable to set the pipeline to playing state!", __func__);
    }
    mIsRecording = true;
    log_info("%s : GST start record, save %s", __func__, path.toLocal8Bit().data());
}

void GstHandle::stopRecord()
{
    if(!mIsRecording){
        return;
    }

    if(gst_element_send_event(mQtMux, gst_event_new_eos()) != TRUE){
        log_error("%s : GST send eos to qtmux error, may mp4 cannot play", __func__);
    }

    gst_element_set_state(mRecordQueue, GST_STATE_NULL);
    gst_element_set_state(mH264Parse, GST_STATE_NULL);
    gst_element_set_state(mQtMux, GST_STATE_NULL);
    gst_element_set_state(mFileSink, GST_STATE_NULL);

    record_sink_pad = gst_element_get_static_pad(mRecordQueue, "sink");

    if(gst_pad_is_linked(record_sink_pad))
        gst_pad_unlink(tee_record_pad, record_sink_pad);
    gst_bin_remove(GST_BIN(mPipeline), mFileSink);
    gst_bin_remove(GST_BIN(mPipeline), mQtMux);
    gst_bin_remove(GST_BIN(mPipeline), mH264Parse);
    gst_bin_remove(GST_BIN(mPipeline), mRecordQueue);
    gst_object_unref(record_sink_pad);
    gst_element_release_request_pad(mTee, tee_record_pad);
    gst_object_unref(tee_record_pad);

    mIsRecording = false;
    log_info("%s : GST stop record", __func__);
}

可能的原因

状态同步逻辑混乱：startRecord中多次重复调用gst_element_sync_state_with_parent，且最后强制重置整个pipeline为PLAYING状态，可能导致元素状态冲突，干扰数据流正常流动。
Pad操作顺序错误：stopRecord中先将元素设置为NULL状态再执行pad解绑，易与上游数据流产生竞态，导致Tee或appsrc内部状态异常。
EOS处理不完整：仅给qtmux发送EOS但未等待处理完成就销毁元素，残留的未处理数据可能阻塞上游，引发appsrc拒绝接收数据。
异常分支资源泄漏：部分错误分支未清理已创建的元素/Pad，多次操作后可能导致资源耗尽，触发未知异常。

解决方案

规范状态同步流程
元素添加到bin后，先完成链接再统一同步状态，无需重复设置整个pipeline的PLAYING状态（原pipeline已处于PLAYING状态）：

// 先链接元素，再同步状态
if(gst_element_link_many(mRecordQueue, mH264Parse, mQtMux, mFileSink, NULL) != TRUE)
{
    log_error("%s : GST video queue link error", __func__);
    // 错误分支清理资源
    gst_object_unref(mFileSink);
    gst_object_unref(mQtMux);
    gst_object_unref(mH264Parse);
    gst_object_unref(mRecordQueue);
    return;
}
// 统一同步状态到父节点（PLAYING）
gst_element_sync_state_with_parent(mRecordQueue);

修正Pad操作顺序
先执行Pad解绑，再设置元素为NULL状态，最后释放Tee的request pad：

record_sink_pad = gst_element_get_static_pad(mRecordQueue, "sink");
if(gst_pad_is_linked(record_sink_pad))
    gst_pad_unlink(tee_record_pad, record_sink_pad);
gst_object_unref(record_sink_pad);

// 先解绑再重置状态
gst_element_set_state(mRecordQueue, GST_STATE_NULL);
gst_element_set_state(mH264Parse, GST_STATE_NULL);
gst_element_set_state(mQtMux, GST_STATE_NULL);
gst_element_set_state(mFileSink, GST_STATE_NULL);

// 释放Tee的request pad
gst_element_release_request_pad(mTee, tee_record_pad);
gst_object_unref(tee_record_pad);

// 从bin移除元素
gst_bin_remove(GST_BIN(mPipeline), mFileSink);
gst_bin_remove(GST_BIN(mPipeline), mQtMux);
gst_bin_remove(GST_BIN(mPipeline), mH264Parse);
gst_bin_remove(GST_BIN(mPipeline), mRecordQueue);

等待EOS处理完成
发送EOS后，通过gst_element_get_state等待qtmux处理完成，再执行后续销毁操作：

gst_element_send_event(mQtMux, gst_event_new_eos());
// 无限等待EOS处理完成
GstStateChangeReturn ret = gst_element_get_state(mQtMux, NULL, NULL, GST_CLOCK_TIME_NONE);
if(ret != GST_STATE_CHANGE_SUCCESS) {
    log_error("%s : GST wait qtmux EOS failed", __func__);
}

完善异常分支资源清理
所有错误分支都要清理已创建的元素和Pad，避免资源泄漏，例如startRecord中Pad链接失败时：

if(ret != GST_PAD_LINK_OK)
{
    log_error("%s : GST pad link error", __func__);
    gst_object_unref(record_sink_pad);
    // 清理已添加到bin的元素
    gst_bin_remove(GST_BIN(mPipeline), mFileSink);
    gst_bin_remove(GST_BIN(mPipeline), mQtMux);
    gst_bin_remove(GST_BIN(mPipeline), mH264Parse);
    gst_bin_remove(GST_BIN(mPipeline), mRecordQueue);
    // 释放元素引用
    gst_object_unref(mFileSink);
    gst_object_unref(mQtMux);
    gst_object_unref(mH264Parse);
    gst_object_unref(mRecordQueue);
    return;
}

替代控制方法

使用Pad阻塞功能：无需动态创建/销毁分支元素，初始化时就将录制分支添加到pipeline并链接，通过gst_pad_set_blocked控制是否让数据流入录制分支。这种方式避免了动态元素操作的状态同步问题，稳定性更高。
封装录制分支为独立Bin：将queue → h264parse → qtmux → filesink封装为一个单独的GstBin，通过添加/移除Bin来控制录制，状态管理更简洁，出错概率更低。

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