Pyqt5 qthread + 信号不工作 + gui 冻结
Pyqt5 qthread + signal not working + gui freeze
我正在尝试使用 imap lib 制作一个邮箱检查器，它在 python、队列和没有 gui 的多线程上工作得很好。
但是当我尝试放置一个 gui 时，我制作的每一个功能都会让 gui 冻结直到完成。
我尝试了各种文档（添加 qthread、signal、cursorr 等）和教程中的很多东西 none 对我有用。
有人可以帮助我了解如何设置文本或将文本附加到 QtextEdit 而 运行 一个函数因为它只有在完成后才能工作。
这是我的代码：
class Checker(QtCore.QThread):
signal = QtCore.pyqtSignal(object)
def __init__(self, lignesmailtocheck):
QtCore.QThread.__init__(self)
self.lignesmailtocheck = lignesmailtocheck
def run(self):
lignemailtocheck = self.lignesmailtocheck.strip()
maillo, passo = lignemailtocheck.split(":",1)
debmail, finmail = maillo.split("@",1)
setimap =["oultook.com:imap-mail.outlook.com", "gmail.com:imap.gmail.com"]
for lignesimaptocheck in sorted(setimap):
ligneimaptocheck = lignesimaptocheck.strip()
fai, imap = ligneimaptocheck.split(":",1)
if finmail == fai:
passo0 = passo.rstrip()
try :
mail = imaplib.IMAP4_SSL(imap)
mail.login(maillo, passo)
mailboxok = open("MailBoxOk.txt", "a+", encoding='utf-8', errors='ignore')
mailboxok.write(maillo+":"+passo+"\n")
mailboxok.close()
totaly = maillo+":"+passo0+":"+imap
print(maillo+":"+passo+"\n")
self.send_text.emit(totaly)
time.sleep(1)
except imaplib.IMAP4.error:
print ("LOGIN FAILED!!! ")
class Ui_Form(object):
def setupUi(self, Form):
Form.setObjectName("Form")
Form.resize(400, 300)
self.pushButton = QtWidgets.QPushButton(Form)
self.pushButton.setGeometry(QtCore.QRect(150, 210, 75, 23))
self.pushButton.setObjectName("pushButton")
self.pushButton.clicked.connect(self.gogogo)
self.openliste = QtWidgets.QToolButton(Form)
self.openliste.setGeometry(QtCore.QRect(40, 110, 71, 21))
self.openliste.setObjectName("openliste")
self.textEdit = QtWidgets.QTextEdit(Form)
self.textEdit.setGeometry(QtCore.QRect(170, 50, 201, 121))
self.textEdit.setObjectName("textEdit")
self.progressBar = QtWidgets.QProgressBar(Form)
self.progressBar.setGeometry(QtCore.QRect(10, 260, 381, 23))
self.progressBar.setValue(0)
self.progressBar.setObjectName("progressBar")
self.retranslateUi(Form)
QtCore.QMetaObject.connectSlotsByName(Form)
def retranslateUi(self, Form):
_translate = QtCore.QCoreApplication.translate
Form.setWindowTitle(_translate("Form", "Form"))
self.pushButton.setText(_translate("Form", "PushButton"))
self.openliste.setText(_translate("Form", "..."))
def gogogo(self):
mailtocheck = open('File/toCheck.txt', 'r', encoding='utf-8', errors='ignore').readlines()
setmailtocheck = set(mailtocheck)
for lignesmailtocheck in sorted(setmailtocheck):
checker = Checker(lignesmailtocheck)
thread = QThread()
checker.moveToThread(thread)
# connections after move so cross-thread:
thread.started.connect(checker.run)
checker.signal.connect(self.checkedok)
thread.start()
def checkedok(self, data):
print(data)
self.textEdit.append(data)
if __name__ == "__main__":
app = QtWidgets.QApplication(sys.argv)
Form = QtWidgets.QWidget()
ui = Ui_Form()
ui.setupUi(Form)
Form.show()
sys.exit(app.exec_())
我无法测试，因为 setimap 在我的系统上不可用。我将 CheckerThread 重命名为 Checker 因为它不再是线程（它只是线程中的 "lives"）：
class Checker(QtCore.QObject):
然后只需将 gogogo(self) 中的循环内容替换为：
for lignesmailtocheck in sorted(setmailtocheck):
checker = Checker(lignesmailtocheck)
thread = QThread()
checker.moveToThread(thread)
# connections after move so cross-thread:
thread.started.connect(checker.run)
checker.signal.connect(self.checkedok)
thread.start()
self.threads.append(thread)
用 pyqtSlot 装饰插槽几乎总是一个好主意，因此 run 和 checkedok 都应该这样装饰。
SO answer about Qt threads 非常方便地提醒您自己的细节（但请注意，它使用旧式连接——您必须将 C++ connect( sender, SIGNAL(sig), receiver, SLOT(slot)); 转换为 PyQt5 sender.sig.connect(receiver.slot)）。
由于在PyQt中经常有关于使用QThread的问题，类似于你的问题，这里有一个例子来说明如何在PyQt中正确使用线程。我希望它可以作为类似问题的 goto-answer 有用，所以我花了比平常更多的时间来准备它。
该示例创建了许多在 non-main 线程中执行并通过 Qt 的异步信号与主（即 GUI）线程通信的工作对象。
import time
import sys
from PyQt5.QtCore import QObject, QThread, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import QApplication, QPushButton, QTextEdit, QVBoxLayout, QWidget
def trap_exc_during_debug(*args):
# when app raises uncaught exception, print info
print(args)
# install exception hook: without this, uncaught exception would cause application to exit
sys.excepthook = trap_exc_during_debug
class Worker(QObject):
"""
Must derive from QObject in order to emit signals, connect slots to other signals, and operate in a QThread.
"""
sig_step = pyqtSignal(int, str) # worker id, step description: emitted every step through work() loop
sig_done = pyqtSignal(int) # worker id: emitted at end of work()
sig_msg = pyqtSignal(str) # message to be shown to user
def __init__(self, id: int):
super().__init__()
self.__id = id
self.__abort = False
@pyqtSlot()
def work(self):
"""
Pretend this worker method does work that takes a long time. During this time, the thread's
event loop is blocked, except if the application's processEvents() is called: this gives every
thread (incl. main) a chance to process events, which in this sample means processing signals
received from GUI (such as abort).
"""
thread_name = QThread.currentThread().objectName()
thread_id = int(QThread.currentThreadId()) # cast to int() is necessary
self.sig_msg.emit('Running worker #{} from thread "{}" (#{})'.format(self.__id, thread_name, thread_id))
for step in range(100):
time.sleep(0.1)
self.sig_step.emit(self.__id, 'step ' + str(step))
# check if we need to abort the loop; need to process events to receive signals;
app.processEvents() # this could cause change to self.__abort
if self.__abort:
# note that "step" value will not necessarily be same for every thread
self.sig_msg.emit('Worker #{} aborting work at step {}'.format(self.__id, step))
break
self.sig_done.emit(self.__id)
def abort(self):
self.sig_msg.emit('Worker #{} notified to abort'.format(self.__id))
self.__abort = True
class MyWidget(QWidget):
NUM_THREADS = 5
# sig_start = pyqtSignal() # needed only due to PyCharm debugger bug (!)
sig_abort_workers = pyqtSignal()
def __init__(self):
super().__init__()
self.setWindowTitle("Thread Example")
form_layout = QVBoxLayout()
self.setLayout(form_layout)
self.resize(400, 800)
self.button_start_threads = QPushButton()
self.button_start_threads.clicked.connect(self.start_threads)
self.button_start_threads.setText("Start {} threads".format(self.NUM_THREADS))
form_layout.addWidget(self.button_start_threads)
self.button_stop_threads = QPushButton()
self.button_stop_threads.clicked.connect(self.abort_workers)
self.button_stop_threads.setText("Stop threads")
self.button_stop_threads.setDisabled(True)
form_layout.addWidget(self.button_stop_threads)
self.log = QTextEdit()
form_layout.addWidget(self.log)
self.progress = QTextEdit()
form_layout.addWidget(self.progress)
QThread.currentThread().setObjectName('main') # threads can be named, useful for log output
self.__workers_done = None
self.__threads = None
def start_threads(self):
self.log.append('starting {} threads'.format(self.NUM_THREADS))
self.button_start_threads.setDisabled(True)
self.button_stop_threads.setEnabled(True)
self.__workers_done = 0
self.__threads = []
for idx in range(self.NUM_THREADS):
worker = Worker(idx)
thread = QThread()
thread.setObjectName('thread_' + str(idx))
self.__threads.append((thread, worker)) # need to store worker too otherwise will be gc'd
worker.moveToThread(thread)
# get progress messages from worker:
worker.sig_step.connect(self.on_worker_step)
worker.sig_done.connect(self.on_worker_done)
worker.sig_msg.connect(self.log.append)
# control worker:
self.sig_abort_workers.connect(worker.abort)
# get read to start worker:
# self.sig_start.connect(worker.work) # needed due to PyCharm debugger bug (!); comment out next line
thread.started.connect(worker.work)
thread.start() # this will emit 'started' and start thread's event loop
# self.sig_start.emit() # needed due to PyCharm debugger bug (!)
@pyqtSlot(int, str)
def on_worker_step(self, worker_id: int, data: str):
self.log.append('Worker #{}: {}'.format(worker_id, data))
self.progress.append('{}: {}'.format(worker_id, data))
@pyqtSlot(int)
def on_worker_done(self, worker_id):
self.log.append('worker #{} done'.format(worker_id))
self.progress.append('-- Worker {} DONE'.format(worker_id))
self.__workers_done += 1
if self.__workers_done == self.NUM_THREADS:
self.log.append('No more workers active')
self.button_start_threads.setEnabled(True)
self.button_stop_threads.setDisabled(True)
# self.__threads = None
@pyqtSlot()
def abort_workers(self):
self.sig_abort_workers.emit()
self.log.append('Asking each worker to abort')
for thread, worker in self.__threads: # note nice unpacking by Python, avoids indexing
thread.quit() # this will quit **as soon as thread event loop unblocks**
thread.wait() # <- so you need to wait for it to *actually* quit
# even though threads have exited, there may still be messages on the main thread's
# queue (messages that threads emitted before the abort):
self.log.append('All threads exited')
if __name__ == "__main__":
app = QApplication([])
form = MyWidget()
form.show()
sys.exit(app.exec_())
理解 multi-thread PyQt 编程所必需的主要概念如下：
- Qt 线程有自己的事件循环（特定于每个线程）。主线程，又名 GUI 线程，也是一个
QThread，它的事件循环由该线程管理。
- 线程之间的信号通过接收线程的事件循环进行（异步）传输。因此，GUI 或任何线程的响应能力 = 处理事件的能力。例如，如果一个线程在一个函数循环中很忙，它就不能处理事件，所以它不会响应来自 GUI 的信号，直到函数 returns。
- 如果线程中的工作对象（方法）可能必须根据来自 GUI 的信号更改其操作过程（例如，中断循环或等待），它必须调用
processEvents() QApplication 实例。这将允许 QThread 处理事件，从而调用槽以响应来自 GUI 的异步信号。请注意，QApplication.instance().processEvents() 似乎在每个线程上调用 processEvents()，如果不需要，则 QThread.currentThread().processEvents() 是一个有效的替代方法。
- 对
QThread.quit() 的调用不会立即退出其事件循环：它必须等待当前正在执行的插槽（如果有）到 return。因此，一旦线程被告知退出，您就必须对其执行 wait() 操作。因此，中止工作线程通常涉及（通过自定义信号）向其发出信号以停止其正在执行的操作：这需要 GUI 对象上的自定义信号，该信号与工作槽的连接，并且工作线程的工作方法必须调用线程的 processEvents() 允许发出的信号在工作时到达插槽。
抱歉回答晚了，但这是一种可以解决类似问题的技术。
问题很清楚了。 GUI 冻结，因为它的线程必须执行另一项工作。
下面给出了一个抽象的（从 PyQt 点）解决方案：
- 创建一个继承自 threading.Thread 的 class，它将成为工人。
- 将队列(queue.Queue)作为通信方式传递给构造函数。
- 您可以从 GUI 线程启动工作线程并使用队列传递消息。
- 要使 GUI 线程读取消息，请创建一个具有您选择的时间间隔的 QTimer 并注册一个回调函数。在回调函数中读取队列。
示例代码：
class Worker(threading.Thread):
def __init__(self, queue):
super().init()
self.queue = queue
def run(self):
# Your code that uses self.queue.put(object)
class Gui:
def __init__(self):
self.timer = Qtimer()
self.timer.setInterval(milliseconds)
self.timer.timeout.connect(self.read_data)
def start_worker(self):
self.queue = queue.Queue()
thr = Worker(self.queue)
thr.start()
def read_data(self):
data = self.queue.get()
self.timer.timeout.connect注册回调函数
我正在尝试使用 imap lib 制作一个邮箱检查器，它在 python、队列和没有 gui 的多线程上工作得很好。
但是当我尝试放置一个 gui 时，我制作的每一个功能都会让 gui 冻结直到完成。
我尝试了各种文档（添加 qthread、signal、cursorr 等）和教程中的很多东西 none 对我有用。
有人可以帮助我了解如何设置文本或将文本附加到 QtextEdit 而 运行 一个函数因为它只有在完成后才能工作。
这是我的代码：
class Checker(QtCore.QThread):
signal = QtCore.pyqtSignal(object)
def __init__(self, lignesmailtocheck):
QtCore.QThread.__init__(self)
self.lignesmailtocheck = lignesmailtocheck
def run(self):
lignemailtocheck = self.lignesmailtocheck.strip()
maillo, passo = lignemailtocheck.split(":",1)
debmail, finmail = maillo.split("@",1)
setimap =["oultook.com:imap-mail.outlook.com", "gmail.com:imap.gmail.com"]
for lignesimaptocheck in sorted(setimap):
ligneimaptocheck = lignesimaptocheck.strip()
fai, imap = ligneimaptocheck.split(":",1)
if finmail == fai:
passo0 = passo.rstrip()
try :
mail = imaplib.IMAP4_SSL(imap)
mail.login(maillo, passo)
mailboxok = open("MailBoxOk.txt", "a+", encoding='utf-8', errors='ignore')
mailboxok.write(maillo+":"+passo+"\n")
mailboxok.close()
totaly = maillo+":"+passo0+":"+imap
print(maillo+":"+passo+"\n")
self.send_text.emit(totaly)
time.sleep(1)
except imaplib.IMAP4.error:
print ("LOGIN FAILED!!! ")
class Ui_Form(object):
def setupUi(self, Form):
Form.setObjectName("Form")
Form.resize(400, 300)
self.pushButton = QtWidgets.QPushButton(Form)
self.pushButton.setGeometry(QtCore.QRect(150, 210, 75, 23))
self.pushButton.setObjectName("pushButton")
self.pushButton.clicked.connect(self.gogogo)
self.openliste = QtWidgets.QToolButton(Form)
self.openliste.setGeometry(QtCore.QRect(40, 110, 71, 21))
self.openliste.setObjectName("openliste")
self.textEdit = QtWidgets.QTextEdit(Form)
self.textEdit.setGeometry(QtCore.QRect(170, 50, 201, 121))
self.textEdit.setObjectName("textEdit")
self.progressBar = QtWidgets.QProgressBar(Form)
self.progressBar.setGeometry(QtCore.QRect(10, 260, 381, 23))
self.progressBar.setValue(0)
self.progressBar.setObjectName("progressBar")
self.retranslateUi(Form)
QtCore.QMetaObject.connectSlotsByName(Form)
def retranslateUi(self, Form):
_translate = QtCore.QCoreApplication.translate
Form.setWindowTitle(_translate("Form", "Form"))
self.pushButton.setText(_translate("Form", "PushButton"))
self.openliste.setText(_translate("Form", "..."))
def gogogo(self):
mailtocheck = open('File/toCheck.txt', 'r', encoding='utf-8', errors='ignore').readlines()
setmailtocheck = set(mailtocheck)
for lignesmailtocheck in sorted(setmailtocheck):
checker = Checker(lignesmailtocheck)
thread = QThread()
checker.moveToThread(thread)
# connections after move so cross-thread:
thread.started.connect(checker.run)
checker.signal.connect(self.checkedok)
thread.start()
def checkedok(self, data):
print(data)
self.textEdit.append(data)
if __name__ == "__main__":
app = QtWidgets.QApplication(sys.argv)
Form = QtWidgets.QWidget()
ui = Ui_Form()
ui.setupUi(Form)
Form.show()
sys.exit(app.exec_())
我无法测试，因为 setimap 在我的系统上不可用。我将 CheckerThread 重命名为 Checker 因为它不再是线程（它只是线程中的 "lives"）：
class Checker(QtCore.QObject):
然后只需将 gogogo(self) 中的循环内容替换为：
for lignesmailtocheck in sorted(setmailtocheck):
checker = Checker(lignesmailtocheck)
thread = QThread()
checker.moveToThread(thread)
# connections after move so cross-thread:
thread.started.connect(checker.run)
checker.signal.connect(self.checkedok)
thread.start()
self.threads.append(thread)
用 pyqtSlot 装饰插槽几乎总是一个好主意，因此 run 和 checkedok 都应该这样装饰。
SO answer about Qt threads 非常方便地提醒您自己的细节（但请注意，它使用旧式连接——您必须将 C++ connect( sender, SIGNAL(sig), receiver, SLOT(slot)); 转换为 PyQt5 sender.sig.connect(receiver.slot)）。
由于在PyQt中经常有关于使用QThread的问题，类似于你的问题，这里有一个例子来说明如何在PyQt中正确使用线程。我希望它可以作为类似问题的 goto-answer 有用，所以我花了比平常更多的时间来准备它。
该示例创建了许多在 non-main 线程中执行并通过 Qt 的异步信号与主（即 GUI）线程通信的工作对象。
import time
import sys
from PyQt5.QtCore import QObject, QThread, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import QApplication, QPushButton, QTextEdit, QVBoxLayout, QWidget
def trap_exc_during_debug(*args):
# when app raises uncaught exception, print info
print(args)
# install exception hook: without this, uncaught exception would cause application to exit
sys.excepthook = trap_exc_during_debug
class Worker(QObject):
"""
Must derive from QObject in order to emit signals, connect slots to other signals, and operate in a QThread.
"""
sig_step = pyqtSignal(int, str) # worker id, step description: emitted every step through work() loop
sig_done = pyqtSignal(int) # worker id: emitted at end of work()
sig_msg = pyqtSignal(str) # message to be shown to user
def __init__(self, id: int):
super().__init__()
self.__id = id
self.__abort = False
@pyqtSlot()
def work(self):
"""
Pretend this worker method does work that takes a long time. During this time, the thread's
event loop is blocked, except if the application's processEvents() is called: this gives every
thread (incl. main) a chance to process events, which in this sample means processing signals
received from GUI (such as abort).
"""
thread_name = QThread.currentThread().objectName()
thread_id = int(QThread.currentThreadId()) # cast to int() is necessary
self.sig_msg.emit('Running worker #{} from thread "{}" (#{})'.format(self.__id, thread_name, thread_id))
for step in range(100):
time.sleep(0.1)
self.sig_step.emit(self.__id, 'step ' + str(step))
# check if we need to abort the loop; need to process events to receive signals;
app.processEvents() # this could cause change to self.__abort
if self.__abort:
# note that "step" value will not necessarily be same for every thread
self.sig_msg.emit('Worker #{} aborting work at step {}'.format(self.__id, step))
break
self.sig_done.emit(self.__id)
def abort(self):
self.sig_msg.emit('Worker #{} notified to abort'.format(self.__id))
self.__abort = True
class MyWidget(QWidget):
NUM_THREADS = 5
# sig_start = pyqtSignal() # needed only due to PyCharm debugger bug (!)
sig_abort_workers = pyqtSignal()
def __init__(self):
super().__init__()
self.setWindowTitle("Thread Example")
form_layout = QVBoxLayout()
self.setLayout(form_layout)
self.resize(400, 800)
self.button_start_threads = QPushButton()
self.button_start_threads.clicked.connect(self.start_threads)
self.button_start_threads.setText("Start {} threads".format(self.NUM_THREADS))
form_layout.addWidget(self.button_start_threads)
self.button_stop_threads = QPushButton()
self.button_stop_threads.clicked.connect(self.abort_workers)
self.button_stop_threads.setText("Stop threads")
self.button_stop_threads.setDisabled(True)
form_layout.addWidget(self.button_stop_threads)
self.log = QTextEdit()
form_layout.addWidget(self.log)
self.progress = QTextEdit()
form_layout.addWidget(self.progress)
QThread.currentThread().setObjectName('main') # threads can be named, useful for log output
self.__workers_done = None
self.__threads = None
def start_threads(self):
self.log.append('starting {} threads'.format(self.NUM_THREADS))
self.button_start_threads.setDisabled(True)
self.button_stop_threads.setEnabled(True)
self.__workers_done = 0
self.__threads = []
for idx in range(self.NUM_THREADS):
worker = Worker(idx)
thread = QThread()
thread.setObjectName('thread_' + str(idx))
self.__threads.append((thread, worker)) # need to store worker too otherwise will be gc'd
worker.moveToThread(thread)
# get progress messages from worker:
worker.sig_step.connect(self.on_worker_step)
worker.sig_done.connect(self.on_worker_done)
worker.sig_msg.connect(self.log.append)
# control worker:
self.sig_abort_workers.connect(worker.abort)
# get read to start worker:
# self.sig_start.connect(worker.work) # needed due to PyCharm debugger bug (!); comment out next line
thread.started.connect(worker.work)
thread.start() # this will emit 'started' and start thread's event loop
# self.sig_start.emit() # needed due to PyCharm debugger bug (!)
@pyqtSlot(int, str)
def on_worker_step(self, worker_id: int, data: str):
self.log.append('Worker #{}: {}'.format(worker_id, data))
self.progress.append('{}: {}'.format(worker_id, data))
@pyqtSlot(int)
def on_worker_done(self, worker_id):
self.log.append('worker #{} done'.format(worker_id))
self.progress.append('-- Worker {} DONE'.format(worker_id))
self.__workers_done += 1
if self.__workers_done == self.NUM_THREADS:
self.log.append('No more workers active')
self.button_start_threads.setEnabled(True)
self.button_stop_threads.setDisabled(True)
# self.__threads = None
@pyqtSlot()
def abort_workers(self):
self.sig_abort_workers.emit()
self.log.append('Asking each worker to abort')
for thread, worker in self.__threads: # note nice unpacking by Python, avoids indexing
thread.quit() # this will quit **as soon as thread event loop unblocks**
thread.wait() # <- so you need to wait for it to *actually* quit
# even though threads have exited, there may still be messages on the main thread's
# queue (messages that threads emitted before the abort):
self.log.append('All threads exited')
if __name__ == "__main__":
app = QApplication([])
form = MyWidget()
form.show()
sys.exit(app.exec_())
理解 multi-thread PyQt 编程所必需的主要概念如下：
- Qt 线程有自己的事件循环（特定于每个线程）。主线程，又名 GUI 线程，也是一个
QThread，它的事件循环由该线程管理。 - 线程之间的信号通过接收线程的事件循环进行（异步）传输。因此，GUI 或任何线程的响应能力 = 处理事件的能力。例如，如果一个线程在一个函数循环中很忙，它就不能处理事件，所以它不会响应来自 GUI 的信号，直到函数 returns。
- 如果线程中的工作对象（方法）可能必须根据来自 GUI 的信号更改其操作过程（例如，中断循环或等待），它必须调用
processEvents()QApplication实例。这将允许 QThread 处理事件，从而调用槽以响应来自 GUI 的异步信号。请注意，QApplication.instance().processEvents()似乎在每个线程上调用processEvents()，如果不需要，则QThread.currentThread().processEvents()是一个有效的替代方法。 - 对
QThread.quit()的调用不会立即退出其事件循环：它必须等待当前正在执行的插槽（如果有）到 return。因此，一旦线程被告知退出，您就必须对其执行 wait() 操作。因此，中止工作线程通常涉及（通过自定义信号）向其发出信号以停止其正在执行的操作：这需要 GUI 对象上的自定义信号，该信号与工作槽的连接，并且工作线程的工作方法必须调用线程的processEvents()允许发出的信号在工作时到达插槽。
抱歉回答晚了，但这是一种可以解决类似问题的技术。
问题很清楚了。 GUI 冻结，因为它的线程必须执行另一项工作。 下面给出了一个抽象的（从 PyQt 点）解决方案：
- 创建一个继承自 threading.Thread 的 class，它将成为工人。
- 将队列(queue.Queue)作为通信方式传递给构造函数。
- 您可以从 GUI 线程启动工作线程并使用队列传递消息。
- 要使 GUI 线程读取消息，请创建一个具有您选择的时间间隔的 QTimer 并注册一个回调函数。在回调函数中读取队列。
示例代码：
class Worker(threading.Thread):
def __init__(self, queue):
super().init()
self.queue = queue
def run(self):
# Your code that uses self.queue.put(object)
class Gui:
def __init__(self):
self.timer = Qtimer()
self.timer.setInterval(milliseconds)
self.timer.timeout.connect(self.read_data)
def start_worker(self):
self.queue = queue.Queue()
thr = Worker(self.queue)
thr.start()
def read_data(self):
data = self.queue.get()
self.timer.timeout.connect注册回调函数