我正在移植一个 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_()
我有两个问题。
我正在为别人回答我自己的问题,因为很长时间没有答案。(我在Qt论坛上发布了同样的问题,但仍然没有答案。见this。)
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)
我们使用上述解决方案大约 6 个月。到目前为止,还没有内存泄漏或性能瓶颈等已知问题。
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))