如何安排函数在 Qt for python 的主 UI 线程上运行?

问题描述 投票:0回答:1

我正在移植一个 python GTK 应用程序以使用 Qt for python(PySide2)。 它使用 python 标准 threading 模块实现工作线程,工作线程使用 Gdk.threads_add_idle() 与主 UI 线程交互。

关于 QThread 的文章很多,但我找不到用 Qt 做这个的简单方法。

我破解并得出如下丑陋的解决方案。 (仅对于核心逻辑,请参阅 IdleRunner 类和 run_on_idle() 函数。)

import sys
import time
import threading
from PySide2.QtCore import *
from PySide2.QtWidgets import *

class IdleRunner(QObject):
    run = Signal(object, tuple, float)
    def __init__(self):
        super().__init__()
        self.run.connect(self.on_run)
    def on_run(self, func, args, delay):
        if delay: QTimer.singleShot(delay * 1000, lambda: func(*args))
        else: func(*args)
_idle_runner = IdleRunner()
def run_on_idle(func, *args, delay = 0):
    _idle_runner.run.emit(func, args, delay)

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__(windowTitle = sys.argv[0])
        self.resize(400, 300)
        self.setAttribute(Qt.WA_DeleteOnClose, True)

        self.label = label = QLabel('aaa', alignment = Qt.AlignCenter)
        self.setCentralWidget(label)

        def thread_entry():
            time.sleep(1)
            run_on_idle(lambda: self.label.setText('bbb'))
            run_on_idle(self.close, delay = 1)
        self.thread = thread = threading.Thread(target = thread_entry)
        thread.start()
    def close(self):
        self.thread.join()
        super().close()

app = QApplication()
main_window = MainWindow()
main_window.show()
app.exec_()

我有两个问题。

  1. 最好的解决方案是什么?
  2. 这个解决方案可能存在的问题是什么? (例如内存泄漏)。
python multithreading qt
1个回答
0
投票

我正在为别人回答我自己的问题,因为很长时间没有答案。(我在Qt论坛上发布了同样的问题,但仍然没有答案。见this。)

  1. 一个好的解决方案似乎是这样的。虽然它不是那么简洁,但它使用适当的 Qt API 处理了两个基本问题。
  • 从后台线程发布事件由 Signal.emit().
  • 在空闲时间运行代码由 QTimer.singleshot().
from PySide2.QtCore import QObject, Signal, QTimer

class IdleRunner(QObject):
    run = Signal(object, tuple, float)
    def __init__(self):
        super().__init__()
        self.run.connect(self.on_run)
    def on_run(self, func, args, delay):
        QTimer.singleShot(delay * 1000, lambda: func(*args))

_idle_runner = IdleRunner()
def run_on_idle(func, *args, delay = 0):
    _idle_runner.run.emit(func, args, delay)

  1. 我们使用上述解决方案大约 6 个月。到目前为止,还没有内存泄漏或性能瓶颈等已知问题。

  2. 9个月后,我找到了一个更有效的解决方案,通过

    QCoreApplication.postEvent()
    从后台线程发布一个事件。 (现在感觉不那么老套了。)下面是支持关键字参数的更扩展的示例。此外,您可以通过将导入语句中的
    PyQt5
    更改为
    PySide2
    来轻松地将其应用于
    PyQt5

from PySide2.QtCore import QObject, QEvent, QTimer, QCoreApplication

class RunEvent(QEvent):
    TYPE = QEvent.Type(QEvent.registerEventType())
    def __init__(self, *args):
        super().__init__(RunEvent.TYPE)
        self.args = args

class IdleRunner(QObject):
    def event(self, e):
        delay, func, args, kwargs = e.args
        if delay == 0:
            func(*args, **kwargs)
        else:
            QTimer.singleShot(int(delay * 1000), lambda: func(*args, **kwargs))
        return True

_idle_runner = IdleRunner()
def run_on_idle(func, *args, delay = 0, **kwargs):
    QCoreApplication.postEvent(_idle_runner, RunEvent(delay, func, args, kwargs))

© www.soinside.com 2019 - 2024. All rights reserved.