Python3线程技术问题：如何从父线程终止Thread A及其所有子线程？

首先得明确：Python的threading.Thread没有原生的强制终止方法——这是有意设计的，因为线程共享进程内存空间，强制终止可能导致锁未释放、数据结构损坏、资源泄漏等严重问题。但既然你明确表示不介意数据丢失，并且需要处理script B阻塞的场景，最优方案是用multiprocessing.Process代替线程，原因如下：

进程拥有独立的内存空间，系统会在进程终止时自动回收其占用的所有资源，不会影响主进程的稳定性；而且Process提供了terminate()方法，可以直接强制杀死进程，包括它创建的子进程（只要做好进程组管理）。

方案1：用multiprocessing.Process替代线程 #

这是最安全、推荐的实现方式，代码示例如下：

import multiprocessing
import time

def execute_script_b():
    # 模拟script B的阻塞/死循环行为
    while True:
        print("Script B is running...")
        time.sleep(1)

if __name__ == "__main__":
    # 为每个script B创建独立进程
    script_process = multiprocessing.Process(target=execute_script_b)
    script_process.start()

    # 模拟运行一段时间后需要终止的场景
    time.sleep(3)
    print("Initiating process termination...")
    # 强制终止进程
    script_process.terminate()
    # 等待进程彻底退出
    script_process.join()
    print("Process terminated successfully.")

处理script B创建子进程的情况 #

如果script B会启动自己的子进程，你可以通过进程组来确保所有子进程一起被终止：

Linux/macOS平台

利用os.setpgid将进程加入新的进程组，然后通过os.killpg杀死整个组：

import multiprocessing
import time
import os
import signal

def execute_script_b():
    # 将当前进程设为新进程组的组长
    os.setpgid(0, 0)
    
    # 模拟script B创建子进程
    def child_task():
        while True:
            print("Child process running...")
            time.sleep(1)
    child_proc = multiprocessing.Process(target=child_task)
    child_proc.start()

    # 模拟script B阻塞
    while True:
        print("Script B is running...")
        time.sleep(1)

if __name__ == "__main__":
    script_process = multiprocessing.Process(target=execute_script_b)
    script_process.start()

    time.sleep(3)
    print("Terminating entire process group...")
    # 获取进程组ID并杀死所有进程
    os.killpg(os.getpgid(script_process.pid), signal.SIGTERM)
    script_process.join()
    print("Process group terminated.")

Windows平台

通过subprocess.CREATE_NEW_PROCESS_GROUP标志创建子进程，这样终止父进程时子进程也会被终止：

import multiprocessing
import time
import subprocess

def execute_script_b():
    # 模拟script B启动外部子进程
    subprocess.Popen(
        "python -c \"while True: print('Child running'); import time; time.sleep(1)\"",
        creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
    )

    # 模拟script B阻塞
    while True:
        print("Script B is running...")
        time.sleep(1)

if __name__ == "__main__":
    script_process = multiprocessing.Process(target=execute_script_b)
    script_process.start()

    time.sleep(3)
    print("Terminating process...")
    script_process.terminate()
    script_process.join()
    print("Process and its children terminated.")

方案2：用底层API强制终止线程（不推荐） #

如果因为某些原因必须用线程，你可以通过ctypes调用系统底层API来终止线程，但风险极高——可能导致GIL锁死、共享数据损坏、资源泄漏等问题，仅作为最后手段：

import threading
import time
import ctypes

def execute_script_b():
    while True:
        print("Script B is running...")
        time.sleep(1)

def force_terminate_thread(thread):
    if not thread.is_alive():
        return
    thread_id = thread.ident
    if hasattr(ctypes, 'windll'):
        # Windows平台：调用TerminateThread
        ctypes.windll.kernel32.TerminateThread(ctypes.c_void_p(thread_id), 0)
    else:
        # Linux/macOS平台：调用pthread_cancel
        libpthread = ctypes.CDLL('libpthread.so.0')
        libpthread.pthread_cancel(thread_id)

if __name__ == "__main__":
    script_thread = threading.Thread(target=execute_script_b)
    script_thread.start()

    time.sleep(3)
    print("Forcing thread termination...")
    force_terminate_thread(script_thread)
    script_thread.join()
    print("Thread terminated.")

总结 #

优先选择进程方案：进程隔离带来的稳定性和资源自动回收特性，完全匹配你"不介意数据丢失、需要强制终止"的需求，是最优解。线程强制终止仅作为极端场景下的备选。

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