在QThread中能否主动抛出KeyboardInterrupt键盘中断？

QThread本身没有原生提供向指定线程抛出特定异常的接口，你可以通过以下两种方案实现需求： #

方案1：Python底层API异步抛出异常 #

这种方法可以直接向目标线程抛出KeyboardInterrupt异常，不需要修改原有业务脚本的逻辑，适配你原有已经做好中断捕获的代码。

实现步骤 #

1. 先在工作对象/线程的运行入口保存当前Python线程的标识

import threading
import ctypes
from PyQt5.QtCore import QObject, QThread, pyqtSignal, QMainWindow

class Worker(QObject):
    finished = pyqtSignal()
    def __init__(self):
        super().__init__()
        self.py_thread_id = None

    def run(self):
        # 存储当前Python线程的ID，后续抛异常需要用到
        self.py_thread_id = threading.get_ident()
        try:
            # 你原有可捕获KeyboardInterrupt的业务脚本
            your_existing_script()
        except KeyboardInterrupt:
            # 你原有处理中断的逻辑
            interrupt_handler()
        finally:
            self.finished.emit()

2. 实现按钮触发的抛异常逻辑

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        # 初始化线程和工作对象
        self.worker = Worker()
        self.thread = QThread()
        self.worker.moveToThread(self.thread)
        self.thread.started.connect(self.worker.run)
        self.worker.finished.connect(self.thread.quit)
        self.worker.finished.connect(self.worker.deleteLater)
        self.thread.finished.connect(self.thread.deleteLater)
        # 绑定取消按钮事件
        self.cancel_button.pressed.connect(self.throw_keyboard_interrupt)
        self.thread.start()

    def throw_keyboard_interrupt(self):
        if not self.worker.py_thread_id:
            return
        # 调用Python底层API向指定线程异步抛出异常
        ctypes.pythonapi.PyThreadState_SetAsyncExc(
            ctypes.c_long(self.worker.py_thread_id),
            ctypes.py_object(KeyboardInterrupt)
        )

注意事项 #

• 仅在线程运行Python代码阶段生效，如果线程卡在Qt C++层的阻塞调用中，异常会等到回到Python解释器执行时才会触发
• 不要向已经退出的线程调用该接口，可能导致程序崩溃

方案2：标志位模拟中断（更稳定） #

这种方式符合Qt多线程开发的最佳实践，没有底层调用的不稳定风险，兼容性更强。
你只需要修改原有业务脚本，在可中断的位置加终止标志位判断，主动抛出KeyboardInterrupt即可：

class Worker(QObject):
    finished = pyqtSignal()
    cancel_flag = False

    def run(self):
        try:
            # 原有业务逻辑的循环/耗时步骤中加标志位判断
            for task in task_list:
                if self.cancel_flag:
                    raise KeyboardInterrupt()
                run_task(task)
        except KeyboardInterrupt:
            # 原有中断处理逻辑
            interrupt_handler()
        finally:
            self.finished.emit()

    # 按钮触发的取消逻辑
    def trigger_cancel(self):
        self.cancel_flag = True

按钮绑定直接调用trigger_cancel方法即可：

self.cancel_button.pressed.connect(self.worker.trigger_cancel)

内容的提问来源于stack exchange，提问作者Rémi Baudoux