摘要
GStreamer框架會自動處理多線程的邏輯，但在某些情況下，我們仍然需要根據實際的情況自己將部分Pipeline在單獨的線程中執行，本文將介紹如何處理這種情況。
GStreamer多線程
GStreamer框架是一個支持多線程的框架，線程會根據Pipeline的需要自動創建和銷毀，例如，將媒體流與應用線程解耦，應用線程不會被GStreamer的處理阻塞。而且，GStreamer的插件還可以創建自己所需的線程用於媒體的處理，例如：在一個4核的CPU上，視頻解碼插件可以創建4個線程來最大化利用CPU資源。
此外，在創建Pipeline時，我們還可以指定某個Pipeline的分支在不同的線程中執行（例如，使audio、video同時在不同的線程中進行解碼）。這是通過queue Element來實現的，queue的sink pad僅僅將數據放入隊列，另外一個線程從隊列中取出數據，並傳遞到下一個Element。queue通常也被用於作為數據緩沖，緩沖區大小可以通過queue的屬性進行配置。
在上面的示例Pipeline中，souce是audiotestsrc，會產生一個相應的audio信號，然后使用tee Element將數據分為兩路，一路被用於播放，通過聲卡輸出，另一路被用於轉換為視頻波形，用於輸出到屏幕。
示例圖中的紅色陰影部分表示位於同一個線程中，queue會創建單獨的線程，所以上面的Pipeline使用了3個線程完成相應的功能。擁有多個sink的Pipeline通常需要多個線程，因為在多個sync間進行同步的時候，sink會阻塞當前所在線程直到所等待的事件發生。
示例代碼
示例代碼將創建上圖所示的Pipeline。
#include <gst/gst.h> int main(int argc, char *argv[]) { GstElement *pipeline, *audio_source, *tee, *audio_queue, *audio_convert, *audio_resample, *audio_sink; GstElement *video_queue, *visual, *video_convert, *video_sink; GstBus *bus; GstMessage *msg; GstPad *tee_audio_pad, *tee_video_pad; GstPad *queue_audio_pad, *queue_video_pad; /* Initialize GStreamer */ gst_init (&argc, &argv); /* Create the elements */ audio_source = gst_element_factory_make ("audiotestsrc", "audio_source"); tee = gst_element_factory_make ("tee", "tee"); audio_queue = gst_element_factory_make ("queue", "audio_queue"); audio_convert = gst_element_factory_make ("audioconvert", "audio_convert"); audio_resample = gst_element_factory_make ("audioresample", "audio_resample"); audio_sink = gst_element_factory_make ("autoaudiosink", "audio_sink"); video_queue = gst_element_factory_make ("queue", "video_queue"); visual = gst_element_factory_make ("wavescope", "visual"); video_convert = gst_element_factory_make ("videoconvert", "csp"); video_sink = gst_element_factory_make ("autovideosink", "video_sink"); /* Create the empty pipeline */ pipeline = gst_pipeline_new ("test-pipeline"); if (!pipeline || !audio_source || !tee || !audio_queue || !audio_convert || !audio_resample || !audio_sink || !video_queue || !visual || !video_convert || !video_sink) { g_printerr ("Not all elements could be created.\n"); return -1; } /* Configure elements */ g_object_set (audio_source, "freq", 215.0f, NULL); g_object_set (visual, "shader", 0, "style", 1, NULL); /* Link all elements that can be automatically linked because they have "Always" pads */ gst_bin_add_many (GST_BIN (pipeline), audio_source, tee, audio_queue, audio_convert, audio_resample, audio_sink, video_queue, visual, video_convert, video_sink, NULL); if (gst_element_link_many (audio_source, tee, NULL) != TRUE || gst_element_link_many (audio_queue, audio_convert, audio_resample, audio_sink, NULL) != TRUE || gst_element_link_many (video_queue, visual, video_convert, video_sink, NULL) != TRUE) { g_printerr ("Elements could not be linked.\n"); gst_object_unref (pipeline); return -1; } /* Manually link the Tee, which has "Request" pads */ tee_audio_pad = gst_element_get_request_pad (tee, "src_%u"); g_print ("Obtained request pad %s for audio branch.\n", gst_pad_get_name (tee_audio_pad)); queue_audio_pad = gst_element_get_static_pad (audio_queue, "sink"); tee_video_pad = gst_element_get_request_pad (tee, "src_%u"); g_print ("Obtained request pad %s for video branch.\n", gst_pad_get_name (tee_video_pad)); queue_video_pad = gst_element_get_static_pad (video_queue, "sink"); if (gst_pad_link (tee_audio_pad, queue_audio_pad) != GST_PAD_LINK_OK || gst_pad_link (tee_video_pad, queue_video_pad) != GST_PAD_LINK_OK) { g_printerr ("Tee could not be linked.\n"); gst_object_unref (pipeline); return -1; } gst_object_unref (queue_audio_pad); gst_object_unref (queue_video_pad); /* Start playing the pipeline */ gst_element_set_state (pipeline, GST_STATE_PLAYING); /* Wait until error or EOS */ bus = gst_element_get_bus (pipeline); msg = gst_bus_timed_pop_filtered (bus, GST_CLOCK_TIME_NONE, GST_MESSAGE_ERROR | GST_MESSAGE_EOS); /* Release the request pads from the Tee, and unref them */ gst_element_release_request_pad (tee, tee_audio_pad); gst_element_release_request_pad (tee, tee_video_pad); gst_object_unref (tee_audio_pad); gst_object_unref (tee_video_pad); /* Free resources */ if (msg != NULL) gst_message_unref (msg); gst_object_unref (bus); gst_element_set_state (pipeline, GST_STATE_NULL); gst_object_unref (pipeline); return 0; }
保存以上代碼，執行下列編譯命令即可得到可執行程序：
gcc basic-tutorial-8.c -o basic-tutorial-8 `pkg-config --cflags --libs gstreamer-1.0`
源碼分析
/* Create the elements */ audio_source = gst_element_factory_make ("audiotestsrc", "audio_source"); tee = gst_element_factory_make ("tee", "tee"); audio_queue = gst_element_factory_make ("queue", "audio_queue"); audio_convert = gst_element_factory_make ("audioconvert", "audio_convert"); audio_resample = gst_element_factory_make ("audioresample", "audio_resample"); audio_sink = gst_element_factory_make ("autoaudiosink", "audio_sink"); video_queue = gst_element_factory_make ("queue", "video_queue"); visual = gst_element_factory_make ("wavescope", "visual"); video_convert = gst_element_factory_make ("videoconvert", "video_convert"); video_sink = gst_element_factory_make ("autovideosink", "video_sink");
首先創建所需的Element：audiotestsrc會產生測試的音頻波形數據。wavescope 會將輸入的音頻數據轉換為波形圖像。audioconvert，audioresample，videoconvert保證了Pipeline中各個Element之間的數據可以互相兼容，使得Pipeline能夠被正確的link起來，如果不需要對數據進行轉換，這些Element會直接將數據發送到下一個Element，這種情況下的性能影響可以忽略不計。
/* Configure elements */ g_object_set (audio_source, "freq", 215.0f, NULL); g_object_set (visual, "shader", 0, "style", 1, NULL);
這里修改相應Element的參數，使得輸出結果更直觀。“freq”會設置audiotestsrc輸出波形的頻率為215Hz，設置“shader”和“style”使得波形更加連續。其他的參數可以通過gst-inspect查看。
/* Link all elements that can be automatically linked because they have "Always" pads */ gst_bin_add_many (GST_BIN (pipeline), audio_source, tee, audio_queue, audio_convert, audio_sink, video_queue, visual, video_convert, video_sink, NULL); if (gst_element_link_many (audio_source, tee, NULL) != TRUE || gst_element_link_many (audio_queue, audio_convert, audio_sink, NULL) != TRUE || gst_element_link_many (video_queue, visual, video_convert, video_sink, NULL) != TRUE) { g_printerr ("Elements could not be linked.\n"); gst_object_unref (pipeline); return -1; }
這里我們使用gst_element_link_many 將多個Element連接起來，需要注意的是，這里我們只連接了擁有Always Pad的Eelement。雖然gst_element_link_many() 能夠在內部處理Request Pad的情況，但我們仍然需要單獨釋放Request Pad，如果直接使用此函數連接所有的Element，這樣容易忘記釋放Request Pad。因此我們使用下面的代碼單獨處理Request Pad。
/* Manually link the Tee, which has "Request" pads */ tee_audio_pad = gst_element_get_request_pad (tee, "src_%u"); g_print ("Obtained request pad %s for audio branch.\n", gst_pad_get_name (tee_audio_pad)); queue_audio_pad = gst_element_get_static_pad (audio_queue, "sink"); tee_video_pad = gst_element_get_request_pad (tee, "src_%u"); g_print ("Obtained request pad %s for video branch.\n", gst_pad_get_name (tee_video_pad)); queue_video_pad = gst_element_get_static_pad (video_queue, "sink"); if (gst_pad_link (tee_audio_pad, queue_audio_pad) != GST_PAD_LINK_OK || gst_pad_link (tee_video_pad, queue_video_pad) != GST_PAD_LINK_OK) { g_printerr ("Tee could not be linked.\n"); gst_object_unref (pipeline); return -1; } gst_object_unref (queue_audio_pad); gst_object_unref (queue_video_pad);
為了能夠連接到Request Pad，我們需要主動的向Element取得相應的Pad。由於一個Element可以提供不同的Request Pad，所以我們需要指定所需的“Pad Template”，Element提供的Pad Template可以通過gst-inspect查看。從下面的結果可以發現，tee提供了2種類型的模板， ”sink“ 和“src_%u"。
$ gst-inspect-1.0 tee ... Pad Templates: SRC template: 'src_%u' Availability: On request Has request_new_pad() function: gst_tee_request_new_pad Capabilities: ANY SINK template: 'sink' Availability: Always Capabilities: ANY ...
由於我們這里需要的是2個Source Pad，所以我們通過gst_element_get_request_pad (tee, "src_%u")獲取兩個Request Pad分別用於audio和video。queue的Sink Pad是Alwasy Pad，所以我們直接使用gst_element_get_static_pad 獲取其Sink Pad。最后再通過gst_pad_link()將其連接起來，在gst_element_link()和gst_element_link_many()內部也是使用此函數連接兩個Element的Pad。
需要注意的是，我們通過Element獲取到的Pad的引用計數會自動增加，因此我們需要調用gst_object_unref()釋放相關的引用，對於Request Pad，我們需要在Pipeline執行完成后進行釋放。
/* Release the request pads from the Tee, and unref them */ gst_element_release_request_pad (tee, tee_audio_pad); gst_element_release_request_pad (tee, tee_video_pad); gst_object_unref (tee_audio_pad); gst_object_unref (tee_video_pad);
除了播放完成后正常的資源釋放外，我們還要對Request進行釋放，需要首先調用gst_element_release_request_pad()，最后再釋放相應的對象。
總結
我們在本文中了解了：
- 如何通過queue讓Pipeline運行在多個線程上。
- 如何通過gst_element_get_request_pad(), gst_pad_link() gst_element_release_request_pad() 對Request Pad進行操作。
- 如何使用tee將一路媒體數據分為多路。