PyQt6线程信号终止失效及线程用法合理性咨询

我有一个无法在处理过程中输出进度的耗时线程操作，因此用另一个线程模拟进度。当耗时操作完成后，发送结束信号通知模拟线程终止，但调用self.thread_two_stop_signal.emit()后，handle_two()仍持续运行，只有断开信号连接后问题才解决。

想了解：

1. 原有方案失效的原因及可行解决方案
2. 该PyQt6线程用法是否正确、能否实现优雅退出
3. 是否存在潜在的线程安全问题

import sys
import time

from PyQt6.QtCore import QObject, pyqtSignal, QThread
from PyQt6.QtWidgets import QApplication, QMainWindow, QProgressBar, QPushButton


class ThreadOne(QObject):
    done_signal = pyqtSignal()
    finished_signal = pyqtSignal()

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

    def run(self):
        for i in range(100):
            time.sleep(0.01)
        self.done_signal.emit()

    def finished(self):
        self.finished_signal.emit()


class ThreadTwo(QObject):
    finished_signal = pyqtSignal()
    progress_signal = pyqtSignal(int)

    def __init__(self):
        self.if_finished = False
        super().__init__()

    def run(self):
        i = 0
        while True:
            if self.if_finished or i == 99:
                self.progress_signal.emit(i)
                return
            i += 1
            self.progress_signal.emit(i)
            time.sleep(0.1)

    def finished(self):
        self.finished_signal.emit()

    def reset(self):
        self.if_finished = False

    def stop(self):
        print("stop")
        self.if_finished = True


class MainWindow(QMainWindow):

    thread_one_do_signal = pyqtSignal()
    thread_one_finished_signal = pyqtSignal()

    thread_two_do_signal = pyqtSignal()
    thread_two_reset_signal = pyqtSignal()
    thread_two_stop_signal = pyqtSignal()
    thread_two_finished_signal = pyqtSignal()

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

        self.setWindowTitle("My PyQt6 App")

        self.setGeometry(100, 100, 400, 200)

        self.btn = QPushButton("Start", self)
        self.btn.setGeometry(150, 25, 50, 50)

        self.bar = QProgressBar(self)
        self.bar.setGeometry(50, 100, 300, 20)

        self.btn.clicked.connect(self.start)

        self.proxy_thread_one = QThread()
        self.thread_one = ThreadOne()
        self.thread_one.moveToThread(self.proxy_thread_one)

        self.thread_one_do_signal.connect(self.thread_one.run)
        self.thread_one_finished_signal.connect(self.thread_one.finished)

        self.thread_one.done_signal.connect(self.handle_one)
        self.thread_one.finished_signal.connect(self.proxy_thread_one.quit)
        self.thread_one.finished_signal.connect(self.proxy_thread_one.wait)

        self.proxy_thread_two = QThread()
        self.thread_two = ThreadTwo()
        self.thread_two.moveToThread(self.proxy_thread_two)

        self.thread_two_do_signal.connect(self.thread_two.run)
        self.thread_two_stop_signal.connect(self.thread_two.stop)
        self.thread_two_reset_signal.connect(self.thread_two.reset)
        self.thread_two_finished_signal.connect(self.thread_two.finished)

        self.thread_two.progress_signal.connect(self.handle_two)
        self.thread_two.finished_signal.connect(self.proxy_thread_two.quit)
        self.thread_two.finished_signal.connect(self.proxy_thread_two.wait)

    def start(self):
        self.proxy_thread_one.start()
        self.proxy_thread_two.start()

        self.thread_one_do_signal.emit()
        self.thread_two_reset_signal.emit()
        self.thread_two_do_signal.emit()

    def handle_one(self):
        self.thread_two_stop_signal.emit()  
        self.thread_two_finished_signal.emit()
        self.thread_one_finished_signal.emit() 
        self.bar.setValue(100)

    def handle_two(self, value):
        self.bar.setValue(value)


if __name__ == "__main__":
    app = QApplication(sys.argv) 
    main_window = MainWindow()  
    main_window.show() 
    sys.exit(app.exec())

def handle_one(self):
        self.thread_two.progress_signal.disconnect(self.handle_two)
        self.thread_two_stop_signal.emit()
        self.thread_two_finished_signal.emit()
        self.thread_one_finished_signal.emit()
        self.bar.setValue(100)

一、原有方案失效原因 #

1. 线程执行时序与信号滞后：当handle_one调用thread_two_stop_signal.emit()时，ThreadTwo的run方法可能正处于time.sleep(0.1)的阻塞状态，if_finished的修改要等sleep结束后才会被检测到。这段等待时间里，ThreadTwo仍会发送progress_signal，而handle_two还保持连接，导致进度条被覆盖回小于100的值。
2. 结束信号触发时机错误：调用stop后立刻触发thread_two_finished_signal，但此时ThreadTwo的run方法还未真正退出。finished_signal触发的proxy_thread_two.quit()不会强制终止正在运行的任务，只会让线程在当前任务完成后退出，所以ThreadTwo仍会继续执行并发送进度信号。

二、可行解决方案 #

方案1：断开信号连接（已采用） #

在通知ThreadTwo停止前，先断开progress_signal与handle_two的连接，后续进度信号不会再更新UI，直接设置进度条为100即可，简单直接。

方案2：等待ThreadTwo真正退出后再处理 #

让ThreadTwo在退出时主动发送结束信号，主线程监听该信号再处理后续逻辑：

• 修改ThreadTwo的run方法：

  def run(self):
      i = 0
      while True:
          if self.if_finished or i == 99:
              self.progress_signal.emit(i)
              self.finished_signal.emit()  # 退出前发送结束信号
              return
          i += 1
          self.progress_signal.emit(i)
          time.sleep(0.1)
  

• 在MainWindow的__init__中添加连接：

  self.thread_two.finished_signal.connect(self.on_thread_two_finished)
  

• 修改handle_one和新增处理方法：

  def handle_one(self):
      self.thread_two_stop_signal.emit()
      self.bar.setValue(100)
  
  def on_thread_two_finished(self):
      self.thread_two_finished_signal.emit()
      self.thread_one_finished_signal.emit()
      self.proxy_thread_two.quit()
      self.proxy_thread_two.wait()
  

方案3：使用线程安全标志位 #

将if_finished改为QAtomicInt，确保跨线程修改的可见性（Python中布尔值修改通常原子性，但原子类型更严谨）：

from PyQt6.QtCore import QAtomicInt

class ThreadTwo(QObject):
    def __init__(self):
        self.if_finished = QAtomicInt(0)
        super().__init__()

    def stop(self):
        print("stop")
        self.if_finished.store(1)

    def run(self):
        i = 0
        while True:
            if self.if_finished.load() == 1 or i == 99:
                self.progress_signal.emit(i)
                self.finished_signal.emit()
                return
            i += 1
            self.progress_signal.emit(i)
            time.sleep(0.1)

三、PyQt6线程用法正确性与优雅退出 #

现有用法的问题 #

1. 重复启动线程：每次点击按钮都会调用start()，但线程只能启动一次，再次调用会报错，需检查线程状态或重新创建线程。
2. 阻塞主线程：__init__中直接将finished_signal连接到proxy_thread_one.wait()，wait()会阻塞主线程导致UI卡顿，应通过监听finished信号处理后续逻辑。
3. 手动触发结束信号不合理：QThread本身自带finished信号，线程事件循环退出时会自动发送，无需手动触发自定义的finished_signal。

优雅退出的正确做法 #

1. 利用QThread原生信号：用QThread.started()触发任务执行，QThread.finished()处理资源清理。
2. 任务完成后自动退出线程：任务函数执行完毕后，线程事件循环自然退出，或调用QThread.quit()。
3. 避免主线程阻塞：不直接调用QThread.wait()，通过finished信号异步处理收尾工作。

优化后的线程初始化示例：

self.proxy_thread_one = QThread()
self.thread_one = ThreadOne()
self.thread_one.moveToThread(self.proxy_thread_one)
# 线程启动自动执行任务
self.proxy_thread_one.started.connect(self.thread_one.run)
# 任务完成后退出线程
self.thread_one.done_signal.connect(self.proxy_thread_one.quit)
# 线程结束后清理资源
self.proxy_thread_one.finished.connect(self.thread_one.deleteLater)
self.proxy_thread_one.finished.connect(self.proxy_thread_one.deleteLater)

四、潜在线程安全问题 #

1. UI操作安全：通过信号槽更新进度条，PyQt跨线程信号默认使用队列连接，UI操作是线程安全的，无问题。
2. 共享变量安全：if_finished布尔值的赋值是原子操作，但复杂共享数据操作需用QMutex保护。
3. 线程重复启动风险：重复调用QThread.start()会报错，需添加状态检查。
4. 资源泄漏：未正确调用quit()和deleteLater()会导致线程和任务对象无法回收，需确保线程结束后清理资源。

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