如何在GStreamer中使用appsrc运行多并行管道及回调处理

你完全不需要为每个管道单独写一个回调函数副本！这其实是GStreamer回调机制设计时就考虑到的场景——利用回调的user_data参数传递每个实例的专属上下文就可以完美解决问题。

核心思路 #

GStreamer的信号回调允许你通过user_data参数传递任意自定义数据，我们可以用一个结构体来存储每个管道实例的专属状态（比如原来的静态变量white、timestamp）和关联元素（appsrc、pipeline），这样同一个回调函数就能区分不同的管道实例，处理各自的数据推送逻辑。

具体实现步骤 #

1. 定义专属上下文结构体

首先创建一个结构体，用来保存每个管道实例的所有必要信息，替换原来的全局静态变量：

typedef struct {
    GstElement *appsrc;       // 当前实例的appsrc元素
    GstElement *pipeline;     // 当前实例的管道
    gboolean white;           // 每个实例独立的黑白切换状态
    GstClockTime timestamp;   // 每个实例独立的时间戳
    gchar *output_path;       // 可选：区分输出文件路径，方便日志排查
} PipelineContext;

2. 修改回调函数，使用上下文数据

把原来依赖全局静态变量的逻辑，改成从user_data中取出上下文结构体，这样每个实例的状态都是独立的：

static void cb_need_data (GstElement *appsrc, guint unused_size, gpointer user_data) {
    // 取出当前管道的上下文
    PipelineContext *ctx = (PipelineContext *)user_data;
    GstBuffer *buffer;
    guint size;
    GstFlowReturn ret;

    size = 1024 * 768 * 2;
    buffer = gst_buffer_new_and_alloc (size);

    // 使用上下文里的独立状态，避免多实例冲突
    memset (buffer->data, ctx->white ? 0x55 : 0xaa, size);
    ctx->white = !ctx->white;

    // 更新当前实例的时间戳
    GST_BUFFER_TIMESTAMP (buffer) = ctx->timestamp;
    GST_BUFFER_DURATION (buffer) = gst_util_uint64_scale_int (1, GST_SECOND, 2);
    ctx->timestamp += GST_BUFFER_DURATION (buffer);

    g_print("push-buffer for pipeline: %s\n", ctx->output_path);

    ret = gst_app_src_push_buffer(GST_APP_SRC(appsrc), buffer);
    if (ret != GST_FLOW_OK) {
        g_print("ret fail for pipeline: %s\n", ctx->output_path);
        g_main_loop_quit (loop);
    }
}

3. 创建多个管道实例并绑定上下文

在main函数中，为每个管道创建对应的上下文结构体，初始化状态后，把它作为user_data传给g_signal_connect，这样回调就能识别当前处理的是哪个实例：

gint main (gint argc, gchar *argv[]) {
    GstElement *pipeline1, *appsrc1;
    GstElement *pipeline2, *appsrc2;
    PipelineContext *ctx1, *ctx2;
    GstCaps *caps;

    // 初始化GStreamer
    gst_init (&argc, &argv);
    loop = g_main_loop_new (NULL, FALSE);

    // ========== 创建第一个管道实例 ==========
    ctx1 = g_malloc0(sizeof(PipelineContext));
    ctx1->output_path = g_strdup("output1.mp4");
    // 构建完整的管道
    pipeline1 = gst_parse_launch(
        "appsrc name=mysource ! queue ! x264enc ! queue ! h264parse ! mp4mux ! filesink location=output1.mp4", 
        NULL
    );
    appsrc1 = gst_bin_get_by_name_recurse_up (GST_BIN (pipeline1), "mysource");
    // 填充上下文
    ctx1->appsrc = appsrc1;
    ctx1->pipeline = pipeline1;
    ctx1->white = FALSE;
    ctx1->timestamp = 0;

    // 设置appsrc的caps
    caps = gst_video_format_new_caps(GST_VIDEO_FORMAT_UYVY, 1024, 768, 0, 1, 4, 3);
    gst_app_src_set_caps(GST_APP_SRC(appsrc1), caps);
    g_object_set (G_OBJECT (appsrc1), "stream-type", 0, "format", GST_FORMAT_TIME, NULL);
    // 绑定回调，传递上下文
    g_signal_connect (appsrc1, "need-data", G_CALLBACK (cb_need_data), ctx1);

    // ========== 创建第二个管道实例 ==========
    ctx2 = g_malloc0(sizeof(PipelineContext));
    ctx2->output_path = g_strdup("output2.mp4");
    pipeline2 = gst_parse_launch(
        "appsrc name=mysource ! queue ! x264enc ! queue ! h264parse ! mp4mux ! filesink location=output2.mp4", 
        NULL
    );
    appsrc2 = gst_bin_get_by_name_recurse_up (GST_BIN (pipeline2), "mysource");
    ctx2->appsrc = appsrc2;
    ctx2->pipeline = pipeline2;
    ctx2->white = FALSE;
    ctx2->timestamp = 0;

    gst_app_src_set_caps(GST_APP_SRC(appsrc2), caps);
    g_object_set (G_OBJECT (appsrc2), "stream-type", 0, "format", GST_FORMAT_TIME, NULL);
    g_signal_connect (appsrc2, "need-data", G_CALLBACK (cb_need_data), ctx2);

    // 启动所有管道
    gst_element_set_state (pipeline1, GST_STATE_PLAYING);
    gst_element_set_state (pipeline2, GST_STATE_PLAYING);
    g_print("ALL PIPELINES PLAYING\n");

    // 运行主循环
    g_main_loop_run (loop);

    // ========== 清理资源 ==========
    gst_element_set_state (pipeline1, GST_STATE_NULL);
    gst_object_unref (GST_OBJECT (pipeline1));
    gst_element_set_state (pipeline2, GST_STATE_NULL);
    gst_object_unref (GST_OBJECT (pipeline2));
    // 释放上下文内存
    g_free(ctx1->output_path);
    g_free(ctx1);
    g_free(ctx2->output_path);
    g_free(ctx2);
    g_main_loop_unref (loop);

    return 0;
}

关键说明 #

• 原来的white和timestamp静态变量是全局共享的，多管道运行时会互相干扰，放到上下文结构体后，每个实例都有自己独立的状态
• 不管你创建多少个管道实例，都只需要一个cb_need_data函数，代码更简洁、易维护
• 如果后续需要扩展管道的属性（比如不同的分辨率、编码参数），只需要修改上下文结构体即可

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