PySide6中QThreadPool线程终止方案咨询：长耗时函数中断难题

方案一：让耗时函数支持可中断（推荐） #

Python线程无法安全强制终止，强行终止可能导致资源泄漏或数据损坏，因此最稳妥的方式是让耗时的example()函数内部定期检查中断信号。

• 用线程安全的threading.Event替代全局status变量，避免多线程竞争问题。
• 在example()的执行流程中，每隔一段逻辑就检查中断信号，一旦触发就提前返回。

代码示例：

import threading
from PySide6.QtCore import QThreadPool, QRunnable

# 线程安全的中断信号，初始为未触发状态
stop_event = threading.Event()

class InferenceTask(QRunnable):
    def run(self):
        for num in num_of_stream:
            if stop_event.is_set():
                break
            self.example()
    
    def example(self):
        # 模拟耗时操作，比如循环处理数据
        for i in range(1000000):
            # 每处理一定量的逻辑就检查中断信号
            if stop_event.is_set():
                return
            # 替换为你的实际耗时业务逻辑
            pass

# 初始化线程池并启动任务
thread_pool = QThreadPool()
task = InferenceTask()
thread_pool.start(task)

# 停止按钮的点击回调
def on_stop_button_clicked():
    stop_event.set()

方案二：用进程替代线程（适配无法修改的第三方耗时函数） #

如果example()是第三方库提供的函数，无法修改其内部逻辑添加中断检查，那么可以用进程来运行任务——进程可以安全终止，不会像线程那样带来风险。

代码示例：

from PySide6.QtCore import QProcess, QObject, Signal

class InferenceProcess(QObject):
    task_finished = Signal()

    def __init__(self):
        super().__init__()
        self.process = QProcess()
    
    def start_task(self):
        # 将耗时逻辑封装到单独的脚本inference_task.py中
        self.process.start("python", ["inference_task.py"])
        self.process.finished.connect(self.task_finished.emit)
    
    def stop_task(self):
        self.process.kill()
        self.process.waitForFinished()

# 使用示例
infer_process = InferenceProcess()
# 启动任务
infer_process.start_task()
# 停止按钮点击回调
def on_stop_button_clicked():
    infer_process.stop_task()

方案三：拆分大任务为小单元 #

把原本单次执行的example()拆分成多个小任务单元，每次只执行一小部分，执行完就回到循环检查中断信号，这样能大幅降低中断响应的延迟。

代码示例：

import threading
from PySide6.QtCore import QThreadPool, QRunnable

stop_event = threading.Event()

def run_inference():
    for num in num_of_stream:
        if stop_event.is_set():
            break
        # 将example的大任务拆分为多个小批次处理
        for batch in split_into_small_batches(your_data):
            if stop_event.is_set():
                return
            process_single_batch(batch)

# 启动任务
thread_pool = QThreadPool()
thread_pool.start(run_inference)

# 停止按钮回调
def on_stop_button_clicked():
    stop_event.set()

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