Python中手动终止子线程与子进程的方法

纸上星河

发布时间：2026-10-12 | 7人浏览过

原创

本文介绍在Python中手动结束子线程和子进程的方法。对于子线程，利用ctypes模块抛出SystemExit异常实现强制终止；对于子进程，直接调用Process对象的terminate()方法。文章提供了完整的代码示例，展示了如何在类中管理多个线程和进程对象，并演示了具体的停止逻辑。

手动终止子线程

在Python中，标准库并没有提供直接强制杀死线程的方法。但可以通过ctypes模块调用底层C API，向目标线程发送异常来实现强制终止。这种方法虽然有效，但属于非正常退出，可能导致资源未正确释放。

核心思路是利用ctypes.pythonapi.PyThreadState_SetAsyncExc函数。该函数可以向指定线程ID发送一个异步异常。如果成功发送，线程在下次执行点会捕获该异常并退出。

import inspect
import ctypes
import threading
from time import sleep
 

def serial_read():
    while True:
        print("春哥纯爷们！")
        sleep(1)
 

def _async_raise(tid, exctype):
    """raises the exception, performs cleanup if needed"""
    tid = ctypes.c_long(tid)
    if not inspect.isclass(exctype):
        exctype = type(exctype)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(exctype))
    if res == 0:
        raise ValueError("invalid thread id")
    elif res != 1:
        # """if it returns a number greater than one, you're in trouble,
        # and you should call it again with exc=NULL to revert the effect"""
        ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, None)
        raise SystemError("PyThreadState_SetAsyncExc failed")
 

def stop_thread(thread):
    _async_raise(thread.ident, SystemExit)
 

def Air():
    ords=0
    myThread = threading.Thread(target=serial_read)
    myThread.start()
    while True:
        ords+=1
        if ords==10:
            stop_thread(myThread)
            print("停止子线程")
            break
        sleep(1)
 

if __name__ == '__main__':
    Air()

在Air函数中，我们创建了一个名为myThread的线程，并启动它。在主循环中，当计数达到10时，调用stop_thread函数，传入线程对象。该函数获取线程ID（thread.ident）并触发SystemExit异常，从而结束线程。

手动终止子进程

与线程不同，子进程是独立的操作系统进程。Python的multiprocessing模块提供了terminate()方法来强制终止子进程。这是更推荐的做法，因为它由操作系统处理进程销毁。

import inspect
import ctypes
from time import sleep
from multiprocessing import Process
 

def serial_read():
    while True:
        print("春哥纯爷们！")
        sleep(1)

def Air():
    ords=0
    myThread = Process(target=serial_read)
    myThread.start()
    while True:
        ords+=1
        if ords==10:
            myThread.terminate()
            print("停止子进程")
            break
        sleep(1)
 

if __name__ == '__main__':
    Air()

在上面的代码中，myThread实际上是一个Process对象。启动后，通过调用myThread.terminate()来强制结束该进程。这种方法简单直接，但同样需要注意进程可能未完全清理资源的情况。

在类中管理多个线程和进程

在实际应用中，我们可能需要同时管理多个线程和进程。可以通过在类中维护一个列表（如among）来存储这些对象，并通过列表索引或遍历来管理它们。

def startss(self,a1,b1,c1,under,rough,blue,among):
    # 创建新线程
    p1=threading.Thread(target=self.line01,args=(a1,b1,under,rough,)) #必须加,号 
    p2=threading.Thread(target=self.line02,args=(a1,c1,under,blue,))
    p3=Process(target=self.Process01,args=(under,rough,blue,)) #计算进程

    #among是类属性list容器
    among.append(p1)
    among.append(p2)
    among.append(p3)

    # 开启新线程
    p1.start()
    p2.start()

    #开启计算用进程
    p3.start()

在startss方法中，我们创建了三个任务对象：p1和p2是线程，p3是进程。这些对象被添加到among列表中，以便后续管理和终止。注意，在传递参数时，如果只有一个参数，也需要加上逗号，例如args=(a1,)，否则会被视为普通变量而非元组。