在QThread中创建GStreamer管道时未收到消息

Question

我有一个PyQt应用程序，当用户按下按钮并在该管道的总线上侦听消息时，该应用程序会创建GStreamer管道。

import gi

gi.require_version("Gst", "1.0")

from gi.repository import Gst, GLib
from PyQt5.QtWidgets import QApplication, QPushButton


Gst.init()

pipeline = None


def on_pipeline_message(bus, message):
    print("Got a message from pipeline:", message.type)
    return True


def on_button_press():
    global pipeline

    pipeline = Gst.parse_launch("videotestsrc ! xvimagesink")
    pipeline.bus.add_watch(GLib.PRIORITY_DEFAULT, on_pipeline_message)
    pipeline.set_state(Gst.State.PLAYING)


app = QApplication([])

playback_button = QPushButton("Press to Start Playback", None)
playback_button.clicked.connect(on_button_press)
playback_button.show()

app.exec()

以上代码按预期工作，并且调用了我的on_pipeline_message回调函数。但是，如果我决定将管道创建代码移到单独的QThread中：

class MakePipelineThread(QThread):
    def run(self):
        global pipeline

        pipeline = Gst.parse_launch("videotestsrc ! xvimagesink")
        pipeline.bus.add_watch(GLib.PRIORITY_DEFAULT, on_pipeline_message)
        pipeline.set_state(Gst.State.PLAYING)

...而是在按下按钮时启动该QThread：

make_pipeline_thread = MakePipelineThread()


def on_button_press():
    make_pipeline_thread.start()

我的on_pipeline_message回调不再运行。如果在单独的QThread中创建管道，为什么有关系？我如何继续接收消息？

Answer 1

GStreamer和Qt都使用GLib.MainContext类型来异步处理发送方和接收方之间的消息。默认情况下¹，GStreamer和Qt都通过全局默认MainContext实例传输消息，可通过GLib.MainContext.default()访问该实例。发送消息时，无论是来自用户输入，来自管道还是其他任何地方的消息，最初都将它们存储在消息队列中。 Qt例行迭代MainContext，它从队列中提取消息并将其发送给任何侦听器。这就是在UI线程上启动管道时能够从GStreamer管道接收消息的原因。

但是，当Qt启动一个新的QThread时，它还会创建一个新的MainContext对象，并将其设置为该线程的默认上下文。当您在QThread中创建GStreamer管道时，您的管道和观察程序会使用该上下文而不是全局默认值进行注册。 Qt不会自动为您迭代QThread的MainContext，因此除非您自己对上下文进行迭代，否则不会收到消息。可以通过在QThread中调用QCoreApplication.processEvents()来完成。

class MakePipelineThread(QThread):
    def run(self):
        global pipeline

        pipeline = Gst.parse_launch("videotestsrc ! xvimagesink")
        pipeline.bus.add_watch(GLib.PRIORITY_DEFAULT, on_pipeline_message)
        pipeline.set_state(Gst.State.PLAYING)

        # Process events until the pipeline reaches the null state
        _, state, _ = pipeline.get_state(Gst.CLOCK_TIME_NONE)
        while state != Gst.State.NULL:
            QCoreApplication.processEvents()
            _, state, _ = pipeline.get_state(Gst.CLOCK_TIME_NONE)

这当然意味着QThread将在管道运行时运行，而不是在管道构建后立即停止。

或者，您可以使用set_sync_handler代替add_watch。这告诉总线在发送消息的同一线程上立即运行回调，而不是通过MainContext异步发送消息。

class MakePipelineThread(QThread):
    def run(self):
        global pipeline

        pipeline = Gst.parse_launch("videotestsrc ! xvimagesink")
        pipeline.bus.set_sync_handler(on_pipeline_message)
        pipeline.set_state(Gst.State.PLAYING)

这消除了完全迭代MainContext的需要，但这意味着您的回调将在GStreamer的“流线程”之一中运行，并在回调运行时阻止该线程执行其他工作。

---

¹某些平台可能在没有GLib支持的情况下编译Qt，在这种情况下，Qt将使用其自己的事件处理系统。在这种情况下，只要应用程序迭代全局默认上下文本身，就不会发生此问题。您可以将QT_NO_GLIB环境变量设置为1，以强制Qt在运行时不使用GLib。