使用GStreamer Appsink取帧后OpenCV imshow显示终端阻塞问题

问题场景 #

用GStreamer的Appsink获取RTSP视频流的每一帧，尝试通过OpenCV的imshow窗口显示，但终端输出一帧信息后就阻塞。试过单独调用imshow、用线程单独运行OpenCV、修改waitKey时长等方法，都无法解决，需要实现OpenCV窗口正常显示视频帧。

问题根源 #

1. UI操作线程冲突：GStreamer的new_sample回调运行在GStreamer内部线程中，直接在回调里调用imshow和waitKey会导致OpenCV的UI线程与GStreamer线程冲突，引发阻塞。
2. 重复初始化GStreamer：每个线程都调用gst_init，不符合GStreamer的初始化规范，应该只在主线程初始化一次。
3. 未明确指定Caps格式：Appsink没有强制指定输出格式，可能导致GStreamer输出的像素格式、步长与OpenCV的Mat不匹配，引发显示异常或阻塞。
4. 锁的不当使用：在回调中用锁包裹UI操作，会导致GStreamer线程被UI操作阻塞，进而阻塞整个管道。

修改后的代码 #

#include <gst/gst.h>
#include <gst/app/gstappsink.h>
#include <string.h>
#include <iostream>
#include <opencv2/opencv.hpp>
#include <vector>
#include <thread>
#include <mutex>
#include <queue>
#include <X11/Xlib.h>

#define CAPS "video/x-raw,format=RGB,pixel-aspect-ratio=1/1"

typedef struct _CustomData {
    GstElement *pipeline;
    GstElement *source;
    GstElement *jitterbuffer;
    GstElement *depayloader;
    GstElement *avdec;
    GstElement *convert;
    GstElement *sink;
    GMainLoop *main_loop;
    std::string rtsp_path;
    std::queue<cv::Mat> frame_queue;
    std::mutex queue_mutex;
    bool running;
} CustomData;

static GstFlowReturn new_sample(GstElement *sink, CustomData *data) {
    GstSample *sample = nullptr;
    GstBuffer *buffer = nullptr;
    GstCaps *caps = nullptr;
    GstMapInfo map;
    gint width, height;

    g_signal_emit_by_name(sink, "pull-sample", &sample);
    if (!sample) {
        return GST_FLOW_ERROR;
    }

    buffer = gst_sample_get_buffer(sample);
    caps = gst_sample_get_caps(sample);
    GstStructure *s = gst_caps_get_structure(caps, 0);
    gst_structure_get_int(s, "width", &width);
    gst_structure_get_int(s, "height", &height);

    if (gst_buffer_map(buffer, &map, GST_MAP_READ)) {
        // 复制帧数据，避免GStreamer释放后数据失效
        cv::Mat frame(height, width, CV_8UC3, map.data).clone();
        
        std::lock_guard<std::mutex> lock(data->queue_mutex);
        if (data->frame_queue.size() > 10) {
            // 限制队列大小，避免内存溢出
            data->frame_queue.pop();
        }
        data->frame_queue.push(frame);

        gst_buffer_unmap(buffer, &map);
    }

    gst_sample_unref(sample);
    return GST_FLOW_OK;
}

static void pad_added_handler(GstElement *src, GstPad *new_pad, CustomData *data) {
    GstPad *sink_pad = gst_element_get_static_pad(data->jitterbuffer, "sink");
    if (gst_pad_is_linked(sink_pad)) {
        gst_object_unref(sink_pad);
        return;
    }

    GstCaps *new_pad_caps = gst_pad_get_current_caps(new_pad);
    GstStructure *new_pad_struct = gst_caps_get_structure(new_pad_caps, 0);
    const gchar *new_pad_type = gst_structure_get_name(new_pad_struct);

    if (!g_str_has_prefix(new_pad_type, "application/x-rtp")) {
        gst_caps_unref(new_pad_caps);
        gst_object_unref(sink_pad);
        return;
    }

    gst_pad_link(new_pad, sink_pad);
    gst_caps_unref(new_pad_caps);
    gst_object_unref(sink_pad);
}

void start_pipeline(CustomData *data) {
    data->pipeline = gst_pipeline_new("pipeline");
    data->source = gst_element_factory_make("rtspsrc", "source");
    data->jitterbuffer = gst_element_factory_make("rtpjitterbuffer", "jitterbuffer");
    data->depayloader = gst_element_factory_make("rtph264depay", "depayloader");
    data->avdec = gst_element_factory_make("avdec_h264", "avdec");
    data->convert = gst_element_factory_make("videoconvert", "convert");
    data->sink = gst_element_factory_make("appsink", "sink");

    g_object_set(data->source, "location", data->rtsp_path.c_str(), NULL);
    // 设置Appsink强制输出指定格式的帧
    GstCaps *caps = gst_caps_from_string(CAPS);
    g_object_set(data->sink, "emit-signals", TRUE, "caps", caps, NULL);
    gst_caps_unref(caps);

    g_signal_connect(data->sink, "new-sample", G_CALLBACK(new_sample), data);
    g_signal_connect(data->source, "pad-added", G_CALLBACK(pad_added_handler), data);

    gst_bin_add_many(GST_BIN(data->pipeline), data->source, data->jitterbuffer, data->depayloader, data->avdec, data->convert, data->sink, NULL);
    gst_element_link_many(data->jitterbuffer, data->depayloader, data->avdec, data->convert, data->sink, NULL);

    data->main_loop = g_main_loop_new(NULL, FALSE);
    gst_element_set_state(data->pipeline, GST_STATE_PLAYING);
    g_main_loop_run(data->main_loop);

    // 停止后标记状态，通知显示线程退出
    data->running = false;
    gst_element_set_state(data->pipeline, GST_STATE_NULL);
    gst_object_unref(GST_OBJECT(data->pipeline));
    g_main_loop_unref(data->main_loop);
}

void display_frames(CustomData *data) {
    data->running = true;
    while (data->running) {
        cv::Mat frame;
        {
            std::lock_guard<std::mutex> lock(data->queue_mutex);
            if (!data->frame_queue.empty()) {
                frame = data->frame_queue.front();
                data->frame_queue.pop();
            }
        }

        if (!frame.empty()) {
            cv::imshow(data->rtsp_path, frame);
            // waitKey必须调用，否则窗口无法刷新
            int key = cv::waitKey(1);
            if (key == 27) { // ESC键触发退出
                g_main_loop_quit(data->main_loop);
                break;
            }
        } else {
            // 无帧时短暂休眠，降低CPU占用
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
    }
    cv::destroyWindow(data->rtsp_path);
}

int main(int argc, char *argv[]) {
    // 统一初始化X11线程和GStreamer
    XInitThreads();
    gst_init(&argc, &argv);

    std::vector<std::string> rtsp_paths = {
        "rtsp://admin:1q2w3e4r@192.168.2.171:561/h264/ch1/av_stream"
    };

    std::vector<std::thread> pipeline_threads;
    std::vector<std::thread> display_threads;
    std::vector<CustomData> data(rtsp_paths.size());

    for (size_t i = 0; i < rtsp_paths.size(); ++i) {
        data[i].rtsp_path = rtsp_paths[i];
        pipeline_threads.emplace_back(start_pipeline, &data[i]);
        display_threads.emplace_back(display_frames, &data[i]);
    }

    for (auto &t : pipeline_threads) {
        t.join();
    }
    for (auto &t : display_threads) {
        t.join();
    }

    return 0;
}

关键修改说明 #

1. 分离UI与GStreamer线程：创建独立的display_frames线程负责OpenCV窗口操作，GStreamer回调仅将帧数据存入线程安全队列，彻底避免线程冲突。
2. 统一初始化：在主线程仅调用一次gst_init和XInitThreads，符合GStreamer的初始化规范。
3. 强制指定输出格式：给Appsink设置明确的RGB格式Caps，确保输出帧与OpenCV的CV_8UC3格式完全匹配，消除格式不兼容问题。
4. 帧数据复制：用clone()复制GStreamer映射的帧数据，避免GStreamer释放缓冲区后数据失效。
5. 队列大小控制：限制帧队列最大长度，防止内存溢出。
6. 优雅退出机制：通过running标志和ESC键实现程序的优雅退出，同时清理GStreamer管道和OpenCV窗口。

编译命令 #

g++ -o gst_opencv gst_opencv.cpp `pkg-config --cflags --libs gstreamer-1.0 gstreamer-app-1.0 opencv4` -pthread -lX11

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