如何无需等待线程完成即可立即终止多线程应用中的所有线程？

先明确：Python本身没有安全且通用的强制线程终止方法，但可以通过以下几种变通方案实现你要的「一键停止所有任务」效果，不用手动维护大量停止标志和嵌套线程状态。

方案1：用进程替代线程实现安全终止 #

把原来的线程任务全部迁移到multiprocessing.Process中运行，不管是QThread还是Python threading的逻辑，都改成进程执行：

• 停止时直接调用process.terminate()，操作系统会强制终止进程，无需关心内部嵌套任务或阻塞操作；
• 进程间通信需用multiprocessing.Queue或Pipe，不能直接共享内存对象；PyQt的GUI操作必须留在主线程，子进程仅处理计算任务，结果通过队列传递给主线程更新UI。

示例代码：

import multiprocessing
import time
from PyQt5.QtWidgets import QPushButton, QWidget, QVBoxLayout

class WorkerProcess(multiprocessing.Process):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue

    def run(self):
        # 替换为你的任务逻辑
        while True:
            self.queue.put("任务运行中...")
            time.sleep(1)

class MainWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.layout = QVBoxLayout()
        self.start_btn = QPushButton("启动任务", self)
        self.stop_btn = QPushButton("停止所有", self)
        self.layout.addWidget(self.start_btn)
        self.layout.addWidget(self.stop_btn)
        self.setLayout(self.layout)

        self.processes = []
        self.queue = multiprocessing.Queue()
        self.start_btn.clicked.connect(self.start_tasks)
        self.stop_btn.clicked.connect(self.stop_all_tasks)

    def start_tasks(self):
        proc = WorkerProcess(self.queue)
        proc.start()
        self.processes.append(proc)

    def stop_all_tasks(self):
        for proc in self.processes:
            if proc.is_alive():
                proc.terminate()
                proc.join()
        self.processes.clear()

方案2：统一管理线程集合，结合Qt中断与应急终止 #

维护一个全局集合，所有启动的QThread和threading.Thread都加入其中，按类型分别处理：

• 对QThread：优先使用Qt内置的requestInterruption()+quit()+wait()，如果线程内有Qt事件循环或可中断阻塞操作（如QThread.sleep()），会自动响应中断；自定义阻塞逻辑只需在关键位置检查self.isInterruptionRequested()，比手动加标志更简洁；
• 对Python threading.Thread：如果上述方法无效，可使用ctypes调用底层API强制终止（极度不推荐，仅作为最后手段），可能导致资源泄漏、死锁。

示例代码：

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

# 全局线程集合
running_threads = set()

class QtWorker(QThread):
    update_signal = pyqtSignal(str)
    def run(self):
        running_threads.add(self)
        while not self.isInterruptionRequested():
            self.update_signal.emit("Qt线程运行中")
            self.msleep(1000)
        running_threads.remove(self)

def py_thread_worker():
    thread = threading.current_thread()
    running_threads.add(thread)
    while True:
        print("Python线程运行中")
        threading.sleep(1)
    running_threads.remove(thread)

# 强制终止Python线程的工具函数（风险高）
def terminate_py_thread(thread):
    if not thread.is_alive():
        return
    tid = thread.ident
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(SystemExit))
    if res > 1:
        ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, 0)

# 一键停止所有任务
def stop_all_threads():
    for thread in list(running_threads):
        if isinstance(thread, QThread):
            thread.requestInterruption()
            thread.quit()
            if not thread.wait(1000):  # 等待1秒，未结束则强制终止
                thread.terminate()
                thread.wait()
            running_threads.remove(thread)
        else:
            terminate_py_thread(thread)
            running_threads.remove(thread)

方案3：用全局停止事件简化标志管理 #

创建一个全局的threading.Event，所有线程在关键循环或阻塞前检查这个事件，无需给每个线程单独加停止标志：

• 优点：仅需维护一个全局事件，代码侵入性低；
• 注意：线程内的阻塞操作需替换为可中断版本，比如用event.wait()代替time.sleep()，否则线程无法及时响应停止信号。

示例代码：

import threading
from PyQt5.QtCore import QThread, pyqtSignal

# 全局停止事件
stop_event = threading.Event()

class QtWorker(QThread):
    update_signal = pyqtSignal(str)
    def run(self):
        while not stop_event.is_set():
            self.update_signal.emit("Qt线程运行中")
            stop_event.wait(1)  # 可中断的1秒等待

def py_thread_worker():
    while not stop_event.is_set():
        print("Python线程运行中")
        stop_event.wait(1)

# 一键停止所有任务
def stop_all_tasks():
    stop_event.set()
    # 等待所有线程退出（可选）
    for thread in threading.enumerate():
        if thread != threading.main_thread():
            thread.join(1)
    stop_event.clear()  # 重置事件，方便后续重新启动

关键注意事项 #

1. 强制终止（进程terminate或线程强制杀死）可能导致资源泄漏（如未关闭的文件、数据库连接），若任务涉及这类资源，最好在任务逻辑中预留清理逻辑；
2. 尽量统一线程模型（比如全用QThread），减少混合两种线程带来的复杂度；
3. 若使用全局停止事件，务必确保线程内的阻塞操作是可中断的，否则无法及时响应停止信号。

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