系列文章目錄

• GStreamer 簡明教程（一）：環境搭建，運行 Basic Tutorial 1 Hello world!
  
• GStreamer 簡明教程（二）：基本概念介紹，Element 和 Pipeline
  
• GStreamer 簡明教程（三）：動態調整 Pipeline
• GStreamer 簡明教程（四）：Seek 以及獲取文件時長
• GStreamer 簡明教程（五）：Pad 相關概念介紹，Pad Capabilities/Templates
• GStreamer 簡明教程（六）：利用 Tee 復制流數據，巧用 Queue 實現多線程
  
• GStreamer 簡明教程（七）：實現管道的動態數據流
• GStreamer 簡明教程（八）：常用工具介紹
• GStreamer 簡明教程（九）：Seek 與跳幀
• GStreamer 簡明教程（十）：插件開發，以一個音頻特效插件為例

---

---

前言

GStreamer 中插件分為三種：Source、Filter 和 Sink，在上一章中我們學習了如何寫一個 Filter 插件，可以說 Filter 插件是最簡單的，因為它只需要關系數據的處理邏輯，而 Source 和 Sink 就更加復雜一些。本章我們來討論如何寫一個 Source 插件。

本章所提及的代碼你可以在 my_plugin 找到。

一、準備工作

準備工作與 GStreamer 簡明教程（十）：插件開發，以一個音頻特效插件為例 中提到的類似，不再贅述。

二、Show me the code

接下來詳細說明代碼中各個細節，其中很多邏輯都是參考 audiotestsrc 來實現的，大家如果想自己對代碼進行詳細的分線，建議寫一個 audiotestsrc 的 demo，進行代碼調試。

3.1 線程模型分析

要理解GStreamer pipeline中的數據流動機制，需要首先明確其線程模型。我們以基礎音頻流水線為例：

audiotestsrc -> autoaudiosink

1. 生產者：audiotestsrc 元素，負責生成音頻測試信號
2. 消費者：autoaudiosink 元素，負責將音頻輸出到系統聲卡

消費端模型

autoaudiosink 的實現通常會依賴系統音頻服務（如ALSA/PulseAudio）的回調機制：

• 系統音頻線程定期通過回調請求數據
• 形成天然的"消費線程"驅動模型

生產端實現方案

生產者有兩種典型的實現范式：

方案A：Push模式（主動生產）

• audiotestsrc 創建獨立的生產者線程
• 持續生成數據并推送(push)至下游
• 下游可能需維護數據緩沖區
• 優勢：實現直接，適合連續數據流
• 缺點：可能需要維護緩沖區

方案B：Pull模式（按需生產）

• audiotestsrc 保持被動狀態
• 當autoaudiosink需要數據時，通過鏈式調用向上游拉取(pull)
• 優勢：流量控制精準
• 難點：需要實現復雜的同步機制

由于Push模式更符合"生產者-消費者"的直觀理解，且實現復雜度較低，本文選擇方案A作為實現基礎。Pull模式涉及GStreamer更底層的調度機制，將在后續深入研究后另文探討。

3.2 深入 audiotestsrc 的實現邏輯

我們先分析官方 audiotestsrc 的關鍵實現，學習 GStreamer 標準 source 元素的調度機制。

3.2.1 生產者線程的啟動時機

當 pipeline 進入播放流程時，audiotestsrc 的生產者線程在 PAUSED 狀態下就已經啟動了。通過調試分析，其線程啟動流程如下：

典型觸發路徑：

1. 狀態切換觸發
   當 gst_element_set_state(pipeline, GST_STATE_PAUSED) 被調用時，會觸發對所有元素的 pad 激活操作：

   gst_pad_set_active(pad, TRUE)  // 激活所有 pad
   

2. Pad 激活回調
   audiotestsrc 的 src pad 重寫了 activatemode_function，此時會調用繼承鏈：

   → gst_base_src_activate_mode()  // GstBaseSrc 的標準實現
   

3. 任務線程創建
   在 gst_base_src_activate_mode() 中，最終通過：

   gst_pad_start_task(pad, gst_base_src_loop, ...)
   

   啟動獨立線程執行主循環邏輯

4. 主循環工作
   gst_base_src_loop() 包含完整處理邏輯：

   • 格式協商（caps negotiation）
   • 發送 STREAM_START 事件
   • 生成音頻數據
   • 數據推送（gst_pad_push()）

關鍵結論：

• 線程啟動的實際觸發點是 PAUSED 狀態下的 pad 激活，而非 PLAYING 狀態
• 通過重寫 activatemode_function 可以自定義啟動邏輯
• GstBaseSrc 已封裝標準線程調度框架，子類只需實現數據生成

---

3.2.2 GStreamer Pad Task 機制詳解

在 GStreamer 框架中，GstPad 不僅負責數據流的連接與協商，還提供了一套完整的異步任務（Task）接口，允許開發者將線程邏輯直接封裝在 Pad 層面，而非傳統的 Element 中。這一設計顯著提升了模塊化程度和靈活性。

核心接口與功能

1. 任務啟動：gst_pad_start_task()

   gboolean gst_pad_start_task(GstPad *pad,GstTaskFunction func,gpointer user_data,GDestroyNotify notify
   );
   

   • 作用：啟動一個專用線程，循環執行指定的 GstTaskFunction。
   • 關鍵特性：
     • 線程會自動進入循環，持續調用目標函數，無需開發者手動實現循環邏輯。
     • 典型應用場景：在 GstBaseSrc 的子類中，gst_base_src_loop() 僅需實現單次數據生成邏輯，任務線程會負責循環調度。例如音頻源（audiosource）可通過此機制持續生成音頻幀。

2. 任務暫停：gst_pad_pause_task()

   • 行為：臨時掛起任務線程的執行，但保留任務狀態（如內部變量）。
   • 用途：實現動態流控，如響應管道的暫停狀態或資源限制。

3. 任務終止：gst_pad_stop_task()

   • 行為：完全停止任務線程并釋放相關資源。
   • 注意：與暫停不同，停止后需重新調用 start_task 才能恢復執行。

設計優勢

• 邏輯解耦：將線程管理與業務邏輯分離，Element 只需關注數據處理，Pad Task 處理線程調度。
• 性能優化：避免在 Element 層頻繁創建/銷毀線程，任務線程可復用。
• 標準化的流控：通過統一的任務接口實現暫停/恢復，簡化狀態管理。

---

3.2.3 loop 中做了哪些事情

在 GStreamer 中，audiotestsrc 這類源元素（source element）通過 gst_pad_start_task 啟動一個任務循環（gst_base_src_loop），其中有兩件事情非常重要

1. 格式協商（negotiate）：
   和下游商量用什么格式傳遞數據（比如采樣率、位深等）。這一步確保數據能被正確處理。

2. 生成數據并推送（push）：
   按協商好的格式生成音頻數據，然后推給下游。

接下來我們重點講 格式協商，當格式確定后如何生成數據和推給下游就會變得簡單很多

3.2.4 格式協商

格式協商的前提是兩個元素已經 link 成功。兩個元素能夠相互連接的前提是它們 pad 的 Capability 是有交集的，比如 src pad 支持的音頻采樣率是 [1, 96000]，那么如果下游支持的采樣率在這個范圍內，他們就能 link 成功，否則在元素 link 階段就會失敗

auto ok = gst_element_link_many(ele0, ele1, NULL);
if(!ok){printf("link failed");
}

在 link 階段僅僅是確認了元素之間支持的數據格式是包含一個子集的，那么接下來在運行階段，我們要從這個子集中，確認唯一的格式，這樣才能確定數據是以什么形式進行流動，例如確定音頻采樣率是 44100，聲道數 2，32位浮點數。也就是說，協商的過程就是找到一個這樣的固定格式，audiotestsrc 根據這個固定格式來生成音頻數據。為了說明這一點，我這邊舉一個簡單的例子。

有三個元素 Source、Filter 和 Sink，它們順序相互連接，支持的采樣率分別是：

• Source : {16000, 32000}
• Filter: [1, Max]
• Sink: {32000, 44100, 48000}

+--------+       +--------+       +--------+
| Source |------>| Filter |------>|  Sink  |
+--------+       +--------+       +--------+
{16000, 32000}   [1, Max]       {16000, 32000, 44100, 48000}

首先，它們的采樣率有一個公共的子集，即 {16000, 32000}，因此它們在 link 階段是成功的；接著，格式協商由 Source 發起，它用自己的格式作為格式過濾器（filter），獲取 peer 端的所支持的格式，流程大致是：

1. Sink 支持采樣率 {16000, 32000, 44100, 48000} 與 {16000, 32000} 取交集，得到 {16000, 32000} 記作 A
2. Filter 支持采樣率 [1, Mac] 與 A 取交集，得到 {16000, 32000} 記作 B
3. Source 支持采樣率 {16000, 32000} 與 B 取交集，得到 {16000, 32000} 記作 peercaps

這時候 peercaps 仍然是一個范圍，不是一個固定的值，最終由 source 來決定使用哪個固定值，固定下來后，再將它作為 source 的 caps，并通過事件通知給其他元素，其他元素會收到 GST_QUERY_ACCEPT_CAPS 事件，在這個事件中獲取 caps 數據做相應的處理。

// 獲取 source 的 caps
thiscaps = gst_pad_query_caps (GST_BASE_SRC_PAD (src), NULL);
// 以 thiscaps 作為 filter，獲取 peercaps
peercaps = gst_pad_peer_query_caps (GST_BASE_SRC_PAD (basesrc), thiscaps);
// basesrc 來決定最終使用哪些固定值
caps = gst_base_src_fixate (basesrc, caps);
// 固定 caps，并發送 GST_QUERY_ACCEPT_CAPS 事件
result = gst_base_src_set_caps (basesrc, caps);

3.2 寫一個 AudioSource 插件

了解了上面的知識后，我們來開始寫一個自己的音頻生成插件，為了讓代碼簡單，我們做了這些簡化：

1. 只支持單聲道、F32LE 、interleave 格式的數據
2. 只支持 push 模式

詳細的代碼實現參考 my_plugin，使用 demo 參考 gstmyaudiotestsrc_example

3.2.1 初始化函數

在類初始化函數如下：

static void gst_my_audio_test_src_class_init(GstMyAudioTestSrcClass *klass) {
//...gobject_class->set_property = gst_my_audio_test_src_set_property;gobject_class->get_property = gst_my_audio_test_src_get_property;gobject_class->finalize = gst_my_audio_test_src_finalize;gst_my_audio_test_class_init_install_properties(gobject_class, klass);
// ...
}

我們覆寫三個函數

1. set_property，用于設置屬性值
2. get_property，用于獲取屬性值
3. finalize，用于類的析構，釋放一些申請的資源

gst_my_audio_test_class_init_install_properties 函數中注冊了屬性，具體大家自行看源碼，不展開說了。

實例的初始化函數如下：

static void gst_my_audio_test_src_init(GstMyAudioTestSrc *filter) {filter->impl = new GstMyAudioTestSrcImpl();filter->srcpad = gst_pad_new_from_static_template(&gst_audio_test_src_src_template, "src");gst_pad_set_activatemode_function(filter->srcpad,gst_my_audio_test_src_activate_mode);gst_element_add_pad(GST_ELEMENT(filter), filter->srcpad);
}

1. 申請 GstMyAudioTestSrcImpl 實例，它用 c++ 來寫，可以簡化一些代碼邏輯
2. 創建 srcpad，并設置 srcpad 的激活函數（_activatemode_function）

3.2.2 激活函數

前面提到，Pad 的 _activatemode_function 是線程啟動的入口，我們看看函數邏輯是怎么樣的

static gboolean gst_my_audio_test_src_activate_mode(GstPad *pad,GstObject *parent,GstPadMode mode,gboolean active) {auto *src = GST_MYAUDIOTESTSRC(parent);switch (mode) {case GST_PAD_MODE_PULL: {res = gst_my_audio_test_src_pull();break;}case GST_PAD_MODE_PUSH: {res = gst_my_audio_test_src_activate_push(src->srcpad, parent, active);break;}// ...
}

目前 GST_PAD_MODE_PULL 是不支持的，因此看 GST_PAD_MODE_PUSH 即可

static gboolean gst_my_audio_test_src_activate_push(GstPad *srcpad,GstObject *parent,gboolean active) {if (active) {g_print("start loop");gst_pad_start_task(srcpad, (GstTaskFunction)gst_my_audio_test_src_loop,srcpad, NULL);} else {g_print("stop loop");gst_pad_stop_task(srcpad);}return TRUE;
}

_activate_push 函數很簡單，啟動 _src_loop 或者停止 _src_loop

3.3.3 loop 循環

接下來看最重要的 _src_loop 函數，主要做的兩個事情就是格式協商和數據填充

格式協商邏輯如下：

static void gst_my_audio_test_src_loop(GstPad *pad) {GstMyAudioTestSrc *src;src = GST_MYAUDIOTESTSRC(GST_OBJECT_PARENT(pad));GstCaps *caps = NULL;gboolean result = FALSE;if (gst_pad_check_reconfigure(pad)) {g_print("need renegotiate\n");GstCaps *thiscaps = gst_pad_query_caps(src->srcpad, NULL);GST_DEBUG_OBJECT(src, "caps of src: %" GST_PTR_FORMAT, thiscaps);GstCaps *peercaps = gst_pad_peer_query_caps(src->srcpad, thiscaps);GST_DEBUG_OBJECT(src, "caps of peer: %" GST_PTR_FORMAT, peercaps);if (peercaps) {caps = peercaps;gst_caps_unref(thiscaps);} else {caps = thiscaps;}if (caps && !gst_caps_is_empty(caps)) {caps = gst_my_audio_test_src_fixate(src, caps);caps = gst_caps_fixate(caps);GST_DEBUG_OBJECT(src, "fixated to: %" GST_PTR_FORMAT, caps);if (gst_caps_is_fixed(caps)) {/* yay, fixed caps, use those then, it's possible that the subclass* does not accept this caps after all and we have to fail. */result = gst_my_audio_test_src_set_caps(src, caps);if (result) {result = gst_pad_push_event(src->srcpad, gst_event_new_caps(caps));}}}if (!result) {GST_DEBUG_OBJECT(src, "negotiation failed");gst_pad_pause_task(pad);}}// ...
}

邏輯大致是：

1. 獲取當前 src pad 的 caps，獲取 peer pad 的 caps，兩者取交集
2. 調用 gst_my_audio_test_src_fixate 去設置 audio src 最期望的數據格式，然后調用 gst_caps_fixate 固化數據格式（此時所有數據格式已經確定，不再是一個范圍值）
3. gst_my_audio_test_src_set_caps 函數從 caps 中獲取 AudioInfo 信息，拿到例如采樣率、聲道數等關鍵信息
4. gst_pad_push_event 發送事件，通知下游數據格式

數據填充數據如下：

static void gst_my_audio_test_src_loop(GstPad *pad) {//...// generate audio dataGstBuffer *buf = NULL;guint blocksize = impl->samples_per_buffer;guint bufferSize = blocksize * sizeof(float);buf = gst_buffer_new_allocate(NULL, bufferSize, NULL);if (buf == NULL) {GST_DEBUG_OBJECT(src, "alloc buffer failed");}GstMapInfo map;gst_buffer_map(buf, &map, GST_MAP_WRITE);float *data = (float *)map.data;impl->fill(data, blocksize);auto ret = gst_pad_push(src->srcpad, buf);if (ret != GST_FLOW_OK) {GST_DEBUG_OBJECT(src, "push buffer failed");gst_pad_pause_task(pad);}
}

1. gst_buffer_new_allocate 申請 GstBuffer，用于存放音頻數據
2. gst_buffer_map 從 GstBuffer 中拿到可寫音頻數據的地址
3. impl->fill(data, blocksize); 用于填充音頻數據，這部分用的是一個 LFO 生成器，具體邏輯大家可以不用在意。總之就是往一塊內存中，寫入生成的音頻數據，你甚至可以寫入隨機噪聲。。
4. gst_pad_push 將數據 push 到下游

總結

以上，我們就將 AudioSource 如何生成數據的邏輯大致講了一遍，運行 gstmyaudiotestsrc_example 之后可以聽到正弦波的聲音。后面我會去研究下如何實現 pull 模式，以及支持更多類型的音頻數據和波形。

參考

• my_plugin
• gstmyaudiotestsrc_example