Pyqt5 qthread + 信号不工作 + gui 冻结

Question

我正在尝试使用imap lib 制作邮箱检查器，它在没有 gui 的情况下与 python、队列和多线程一起工作得很好。

但是当我尝试放置一个 gui 时，我所做的每一个功能，都让 gui 冻结直到完成。

我从各种文档（添加 qthread、signal、cursor 等）和教程中尝试了很多东西，但对我没有用。

有人可以帮助我了解如何在运行函数时将文本设置或附加到 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_())

Answer 1

由于在 PyQt 中使用 QThread 时经常遇到问题，与您的问题类似，这里有一个示例说明如何在 PyQt 中正确使用线程。我希望它可以作为类似问题的转到答案有用，所以我花了比平时更多的时间来准备这个。

该示例创建了许多在非主线程中执行的工作对象，并通过 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_())

理解 PyQt 中多线程编程所必需的主要概念如下：

• Qt 线程有自己的事件循环（特定于每个线程）。主线程，即 GUI 线程，也是 QThread，其事件循环由该线程管理。
• 线程之间的信号通过接收线程的事件循环传输（异步）。因此，GUI 或任何线程的响应能力 = 处理事件的能力。例如，如果一个线程在函数循环中忙，它就不能处理事件，所以在函数返回之前它不会响应来自 GUI 的信号。
• 如果线程中的工作对象（方法）可能必须根据来自 GUI 的信号更改其操作过程（例如，中断循环或等待），它必须在 QApplication 上调用 processEvents()实例。这将允许 QThread 处理事件，从而调用槽以响应来自 GUI 的异步信号。请注意，QApplication.instance().processEvents() 似乎在每个线程上都调用 processEvents()，如果不需要，那么 QThread.currentThread().processEvents() 是一个有效的替代方案。
• 对QThread.quit() 的调用不会立即退出其事件循环：它必须等待当前正在执行的槽（如果有）返回。因此，一旦一个线程被告知要退出，你必须在它上面 wait()。因此，中止工作线程通常涉及（通过自定义信号）发出信号以停止它正在执行的任何操作：这需要 GUI 对象上的自定义信号，将该信号连接到工作插槽，并且工作工作方法必须调用线程的 @ 987654330@ 允许发射的信号在工作时到达插槽。

Answer 2

我无法测试，因为我的系统上没有 setimap。我将CheckerThread 重命名为Checker，因为它不再是一个线程（它只是“生活”在一个线程中）：

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)）。

Answer 3

很抱歉回答迟了，但这是一种可以解决类似问题的技术。

问题很清楚。 GUI 冻结，因为它的线程必须做另一项工作。 下面给出了一个抽象的（从 PyQt 点）解决方案：

1. 创建一个继承自 threading.Thread 的类，该类将成为工作线程。
2. 将队列（queue.Queue）作为通信方式传递给构造函数。
3. 您可以从 GUI 线程启动工作线程并使用队列传递消息。
4. 要让 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 注册回调函数。