PyQt6中使用moveToThread()时QThread的wait()方法异常问题

问题描述 #

尝试使用PyQt6的moveToThread()方式编写多线程程序，创建两个工作线程后调用wait()方法等待线程完成，但程序出现崩溃。已知子类化QThread的方式可以实现需求，但想了解这种写法崩溃的原因。

示例代码：

import sys
import logging
from functools import partial
from PyQt6.QtWidgets import *
from PyQt6.QtCore import *

logging.basicConfig(format="%(message)s", level=logging.INFO)

class MyWork(QObject):
  finished = pyqtSignal()
  def run_work(self, obj_name):
    # sleep 3 secs and finish
    for i in range(3):
      QThread.sleep(1)
      logging.info(f'{obj_name}: {i} sec.')
    self.finished.emit()

class MyWindow(QWidget):
  def __init__(self):
    super().__init__()
    self.setWindowTitle('MyWork Example')

app = QApplication(sys.argv)
form = MyWindow()
form.show()

# create two threads
th1 = QThread()
wo1 = MyWork()
wo1.moveToThread(th1)
th1.started.connect(partial(wo1.run_work, 'obj1'))
wo1.finished.connect(th1.quit)
th1.start()

th2 = QThread()
wo2 = MyWork()
wo2.moveToThread(th2)
th2.started.connect(partial(wo2.run_work, 'obj2'))
wo2.finished.connect(th2.quit)
th2.start()

# join two threads and finish
th1.wait()
th2.wait()
logging.info('All threads finished.')

sys.exit(app.exec())

运行输出：

obj1: 0 sec.
obj2: 0 sec.
obj2: 1 sec.
obj1: 1 sec.
obj2: 2 sec.
obj1: 2 sec.

崩溃原因 #

• 主线程阻塞导致事件循环无法运行：th1.wait()和th2.wait()是阻塞调用，直接卡住了主线程，而Qt的核心事件循环app.exec()还未启动。跨线程的信号（如wo1.finished）需要主线程的事件循环来调度处理，此时主线程被冻结，无法处理这些信号，导致th1.quit()和th2.quit()永远不会被执行，线程无法正常退出，最终引发崩溃。
• 违背moveToThread模式的设计逻辑：moveToThread的核心是通过信号槽实现异步通信来管理线程生命周期，而wait()是同步阻塞方法，强行用同步逻辑干扰异步机制，破坏了Qt的线程调度规则。

正确实现方式 #

通过监听线程的finished信号来追踪所有线程的完成状态，在事件循环正常运行的前提下执行收尾逻辑：

import sys
import logging
from functools import partial
from PyQt6.QtWidgets import *
from PyQt6.QtCore import *

logging.basicConfig(format="%(message)s", level=logging.INFO)

class MyWork(QObject):
    finished = pyqtSignal()
    def run_work(self, obj_name):
        for i in range(3):
            QThread.sleep(1)
            logging.info(f'{obj_name}: {i} sec.')
        self.finished.emit()

class MyWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle('MyWork Example')
        self.completed_threads = 0  # 记录已完成的线程数

    def on_thread_finished(self):
        self.completed_threads += 1
        if self.completed_threads == 2:
            logging.info('All threads finished.')
            QApplication.instance().quit()  # 所有线程完成后退出事件循环

app = QApplication(sys.argv)
form = MyWindow()
form.show()

# 初始化第一个线程
th1 = QThread()
wo1 = MyWork()
wo1.moveToThread(th1)
th1.started.connect(partial(wo1.run_work, 'obj1'))
wo1.finished.connect(th1.quit)
th1.finished.connect(form.on_thread_finished)
th1.start()

# 初始化第二个线程
th2 = QThread()
wo2 = MyWork()
wo2.moveToThread(th2)
th2.started.connect(partial(wo2.run_work, 'obj2'))
wo2.finished.connect(th2.quit)
th2.finished.connect(form.on_thread_finished)
th2.start()

sys.exit(app.exec())

核心要点 #

• 始终让app.exec()处于主线程的最外层，保证事件循环正常运行，这样跨线程信号才能被正确处理。
• 用信号槽替代wait()来实现线程完成后的逻辑，符合Qt异步编程的设计理念，避免主线程阻塞。

内容的提问来源于stack exchange，提问作者Ciphero Chen