python - 尽管使用线程，PyQt5 窗口仍然卡住

Question

为了让我的同事提高安全意识，并向他们展示使用最简单的编码破解密码是多么容易，我编写了一个简单的程序，它只是强力检查每种可能的组合。 为了给他们一个可以使用的小工具，我想将其包装在一个小的 pyqt 应用程序中。 在 Spyder 和控制台上，即使密码较长，破解程序也能正常运行。然而，当使用 PyQT5 时，当我尝试测试更长的密码时，它会卡住。例如。在控制台上破解 abcd34 大约需要 1:30 分钟，使用 PyQt 即使 10 分钟后什么也没有发生。 我关注了这篇文章( Pyqt5 qthread + signal not working + gui freeze )并使用了线程，但它一直卡住。有什么想法我做错了吗？

谢谢

安雅

这是我的代码:

import itertools
import string
import datetime as dt
from PyQt5.QtCore import QObject, QThread, pyqtSignal, pyqtSlot
from PyQt5.QtWidgets import QApplication, QPushButton, QTextEdit, QVBoxLayout, QWidget, QLineEdit


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_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, value):
        """
        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('Starting brute force password cracking.\n\n')

        time1 = dt.datetime.now()
        special = "@%-*,'&`._!#$§äöü?=+:;/\[]{}()"
        chars = string.ascii_letters + string.digits + special
        attempts = 0

        for password_length in range(1, 10):
            for guess in itertools.product(chars, repeat=password_length):
                attempts += 1
                guess = ''.join(guess)
                if guess == value: #self.freetextle.text():
                    time2 = (dt.datetime.now() - time1)
                    self.sig_done.emit(self.__id)
                    return self.sig_msg.emit('Password is {} and it was found in {} guesses, which took me {} seconds'.format(guess, attempts, time2))

        # 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('aborting')

        self.sig_done.emit(self.__id)

    def abort(self):
        self.sig_msg.emit('notified to abort')
        self.__abort = True


class MyWidget(QWidget):

    # sig_start = pyqtSignal()  # needed only due to PyCharm debugger bug (!)
    sig_abort_workers = pyqtSignal()

    def __init__(self):
        super().__init__()

        self.setWindowTitle("Simple password cracker")
        form_layout = QVBoxLayout()
        self.setLayout(form_layout)
        self.resize(400, 400)

        self.button_start_threads = QPushButton()
        self.button_start_threads.clicked.connect(self.start_threads)
        self.button_start_threads.setText("Start")
        form_layout.addWidget(self.button_start_threads)

        # create the input field
        self.freetextle = QLineEdit()
        self.freetextle.setObjectName("password")
        self.freetextle.setText("Enter a password to check")
        form_layout.addWidget(self.freetextle)

        self.log = QTextEdit()
        form_layout.addWidget(self.log)

        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')
        self.button_start_threads.setDisabled(True)

        self.__workers_done = 0
        self.__threads = []
        worker = Worker(1)
        thread = QThread()
        thread.setObjectName('thread_1')# + 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(self.freetextle.text()))
        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))

    @pyqtSlot(int)
    def on_worker_done(self, worker_id):
        self.log.append('done')
        self.__workers_done += 1
        #if self.__workers_done == 1:
        #    self.log.append('No more workers active')
        self.button_start_threads.setEnabled(True)
            # self.__threads = None

    @pyqtSlot()
    def abort_workers(self):
        self.sig_abort_workers.emit()
        self.log.append('Asking 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_())

Answer 1

您的问题出在 thread.started.connect 中。您可以通过传递参数来调用工作方法，而不仅仅是建立信号/槽连接。

鉴于started没有参数，您无法通过此连接将参数传递给插槽。因此，您需要将文本作为参数传递给 worker 构造函数。并在 Worker 中以成员身份使用它。