运行作为 QThread 的繁重的 QTimer 任务

Question

我有一项繁重的任务，每 500 毫秒不断运行s。它包括更新 GUI 元素，我需要随时访问它的变量。

执行的任务：一个动态更新的列表，每 500 毫秒，一个循环遍历该列表并对其中包含的元素执行任务。有时我没有元素，有时我有很多。

加载列表后，用户开始遇到鼠标移动、按键等方面的延迟。毫无疑问，这是因为每 500 毫秒执行一次繁重的任务。

我是否可以将此 QTimer 任务放入 QThread 并不断访问它的元素以更新其中包含的列表？

换句话说，我希望它始终运行在后台运行，但也能够随时更新其中使用的列表。

我正在使用 PySide2；我看过示例，但 none 符合我想要完成的目标。

示例：我想根据需要从主线程更新“aList”元素。如果列表为空，则 for 循环不执行任何操作。否则，它循环遍历元素并将它们加 1。

“运行”函数应设置 500 毫秒的 Qtimer。

有时列表可能是空的，有时又充满了元素。它的大小由 GUI 线程控制。

 from PySide2 import QtCore
 from PySide2 import QtGui 
 from PySide2 import QtWidgets

 import sys
 import time

 class RxProcess(QtCore.QThread):

     output = QtCore.Signal()

     def __init__(self, parent = None):
         super(RxProcess, self).__init__(parent)
         self.aList = list()

    
     def run(self):
    
         # Loop through list
         for element in self.aList: 
        
             element += 1

             # Print to the gui the element that was just updated in the list
             self.output.emit(element)

Answer 1

使用 QThread 很难实现该逻辑（您将不得不使用 QThread.msleep、互斥锁等）。相反，一个简单的解决方案是每 T 秒创建一个新线程，这将使用 threading.Thread + QTimer 实现（也可以使用 QThreadPool + QRunnable + QTimer 实现）：

import random
import sys
import threading
import time


from PySide2 import QtCore, QtWidgets
import shiboken2


class Worker(QtCore.QObject):
    output = QtCore.Signal(object)


def long_running_function(values, worker):
    for element in values:
        time.sleep(0.1)
        if shiboken2.isValid(worker):
            worker.output.emit(element)


class Widget(QtWidgets.QWidget):
    def __init__(self, parent=None):
        super().__init__(parent)

        self.label = QtWidgets.QLabel(alignment=QtCore.Qt.AlignCenter)
        self.button = QtWidgets.QPushButton("Start")

        lay = QtWidgets.QVBoxLayout(self)
        lay.addWidget(self.button)
        lay.addWidget(self.label)

        self.timer = QtCore.QTimer(interval=500)

        self.button.clicked.connect(self.handle_clicked)
        self.timer.timeout.connect(self.launch_task)

    def handle_clicked(self):
        if self.button.text() == "Start":
            self.timer.start()
            self.button.setText("Stop")
        elif self.button.text() == "Stop":
            self.timer.stop()
            self.button.setText("Start")

    def launch_task(self):
        values = random.sample(range(1, 50), 20)
        worker = Worker()
        worker.output.connect(self.label.setNum)
        threading.Thread(
            target=long_running_function,
            args=(values, worker),
            daemon=True,
        ).start()


if __name__ == "__main__":
    app = QtWidgets.QApplication(sys.argv)
    w = Widget()
    w.show()
    sys.exit(app.exec_())

运行作为 QThread 的繁重的 QTimer 任务

Running a heavy QTimer task as a QThread

python

pyside2

qthread

qtimer

运行 作为 QThread 的繁重的 QTimer 任务

Running a heavy QTimer task as a QThread

python

pyside2

qthread

qtimer

运行作为 QThread 的繁重的 QTimer 任务