如何用GStreamer代码监控OpenCV创建的GStreamer管道并解决阻塞问题？

问题场景 #

使用OpenCV 4.6.0结合GStreamer 1.0在IMX8QM开发板上读取UDP流，代码如下：

#define UDP_URL "udpsrc port=15004 buffer-size=5000000 ! watchdog timeout=1000 ! tsdemux latency=0 ! h264parse ! v4l2h264dec ! imxvideoconvert_g2d ! video/x-raw,format=BGRA,width=1280,height=960 ! appsink max-buffers=2"
int main()
{
    cv::VideoCapture video;
    cv::Mat frame;
    video.open(UDP_URL, cv::CAP_GSTREAMER);
    if (!video.isOpened()) {
        printf("Error in opening.\n");
        return -1;
    }

    while(1) {
        if(video.read(frame))
        {
            // 对frame的操作逻辑
        }
        else
            break;
    }

    video.release();
    return 0;
}

遇到的问题：

• 当15004端口无数据传入时，video.read(frame)会阻塞，尤其是v4l2h264dec解码器出现卡顿
• 恢复数据传入后，该函数无法恢复正常，GStreamer管道处于halt状态，即使tcpdump确认数据已到达端口，问题仍存在
• 需要通过GStreamer代码监控管道，但无法访问OpenCV后端的GStreamer组件

解决方案 #

1. 直接构建GStreamer管道，绕过OpenCV封装 #

OpenCV的VideoCapture对GStreamer的封装属于黑盒，无法直接操作管道元素做监控和恢复。直接用GStreamer API构建管道，通过appsink回调获取帧数据传给OpenCV处理，能完全掌控管道状态。

示例代码框架：

#include <opencv2/opencv.hpp>
#include <gst/gst.h>
#include <gst/app/gstappsink.h>

static gboolean new_sample(GstAppSink *sink, gpointer user_data) {
    cv::Mat* frame_ptr = static_cast<cv::Mat*>(user_data);
    GstSample *sample = gst_app_sink_pull_sample(sink);
    if (!sample) return FALSE;

    GstBuffer *buffer = gst_sample_get_buffer(sample);
    GstMapInfo map;
    if (gst_buffer_map(buffer, &map, GST_MAP_READ)) {
        // 根据BGRA格式创建Mat
        *frame_ptr = cv::Mat(960, 1280, CV_8UC4, map.data, map.size / (960 * 4));
        gst_buffer_unmap(buffer, &map);
    }

    gst_sample_unref(sample);
    return TRUE;
}

int main(int argc, char* argv[]) {
    gst_init(&argc, &argv);
    cv::Mat frame;

    // 构建GStreamer管道
    GstElement *pipeline = gst_parse_launch(
        "udpsrc port=15004 buffer-size=5000000 ! watchdog timeout=1000 ! tsdemux latency=0 ! h264parse ! v4l2h264dec ! imxvideoconvert_g2d ! video/x-raw,format=BGRA,width=1280,height=960 ! appsink name=sink max-buffers=2",
        nullptr
    );
    GstElement *appsink = gst_bin_get_by_name(GST_BIN(pipeline), "sink");
    gst_app_sink_set_callbacks(GST_APP_SINK(appsink),
        gst_app_sink_callbacks_new(new_sample, nullptr, nullptr, nullptr, nullptr),
        &frame, nullptr
    );

    gst_element_set_state(pipeline, GST_STATE_PLAYING);

    // 监控管道状态+处理帧
    GstBus *bus = gst_element_get_bus(pipeline);
    while (true) {
        GstMessage *msg = gst_bus_timed_pop(bus, GST_CLOCK_TIME_NONE);
        if (!msg) break;

        switch (GST_MESSAGE_TYPE(msg)) {
            case GST_MESSAGE_ERROR: {
                GError *err;
                gchar *debug;
                gst_message_parse_error(msg, &err, &debug);
                g_printerr("Error: %s\n", err->message);
                g_error_free(err);
                g_free(debug);
                // 尝试重启管道
                gst_element_set_state(pipeline, GST_STATE_NULL);
                gst_element_set_state(pipeline, GST_STATE_PLAYING);
                break;
            }
            case GST_MESSAGE_EOS:
                // 流结束，尝试重启管道
                gst_element_set_state(pipeline, GST_STATE_NULL);
                gst_element_set_state(pipeline, GST_STATE_PLAYING);
                break;
            case GST_MESSAGE_STATE_CHANGED:
                // 监控管道状态变化，判断是否进入halt
                GstState old_state, new_state, pending_state;
                gst_message_parse_state_changed(msg, &old_state, &new_state, &pending_state);
                if (GST_ELEMENT(msg->src) == pipeline) {
                    g_print("Pipeline state changed: %s -> %s\n",
                        gst_element_state_get_name(old_state),
                        gst_element_state_get_name(new_state)
                    );
                }
                break;
            default:
                break;
        }
        gst_message_unref(msg);

        // 处理有效帧
        if (!frame.empty()) {
            // 你的帧处理逻辑
            frame.release();
        }
    }

    gst_object_unref(bus);
    gst_element_set_state(pipeline, GST_STATE_NULL);
    gst_object_unref(pipeline);
    return 0;
}

2. 优化原有GStreamer管道参数 #

• 给v4l2h264dec添加disable-error-checks=false，让解码器在断流时抛出错误而非卡顿
• 给tsdemux设置timeout=1000，让解复用器在无数据时更快触发EOS或错误信号
• 给appsink添加emit-signals=true，配合OpenCV属性提升可控性

修改后的管道字符串：

udpsrc port=15004 buffer-size=5000000 ! watchdog timeout=1000 ! tsdemux timeout=1000 latency=0 ! h264parse ! v4l2h264dec disable-error-checks=false ! imxvideoconvert_g2d ! video/x-raw,format=BGRA,width=1280,height=960 ! appsink max-buffers=2 emit-signals=true

3. 给OpenCV的read操作添加超时机制 #

用线程异步执行video.read，设置超时时间，超时后尝试重新打开VideoCapture：

while(1) {
    bool read_success = false;
    // 异步执行read，避免主线程阻塞
    std::thread read_thread([&](){
        read_success = video.read(frame);
    });
    // 设置5秒超时
    if (read_thread.timed_join(std::chrono::seconds(5))) {
        if(read_success) {
            // 处理帧
        } else {
            break;
        }
    } else {
        // 超时后重启VideoCapture
        video.release();
        if (!video.open(UDP_URL, cv::CAP_GSTREAMER)) {
            printf("Reopen failed.\n");
            return -1;
        }
    }
}

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