Python终止线程的方法

亲测使用如下方法有效，但是如果线程中涉及获取释放锁，可能会导致死锁。

def _async_raise(tid, exctype):
    """
    线程退出，这种方法是强制杀死线程，但是如果线程中涉及获取释放锁，可能会导致死锁。
    :param tid: thread id
    :param exctype: https://docs.python.org/zh-cn/3.8/library/exceptions.html
    :return: None
    """
    """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):
    """
    线程退出封装方法
    :param thread: 线程对象
    :return: None
    """
    _async_raise(thread.ident, SystemExit)

完整示例代码如下：

import time
import ctypes
import inspect
import threading


def _async_raise(tid, exctype):
    """
    线程退出，这种方法是强制杀死线程，但是如果线程中涉及获取释放锁，可能会导致死锁。
    :param tid: thread id
    :param exctype: https://docs.python.org/zh-cn/3.8/library/exceptions.html
    :return: None
    """
    """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):
    """
    线程退出封装方法
    :param thread: 线程对象
    :return: None
    """
    _async_raise(thread.ident, SystemExit)


def run():
    while True:
        print(f"thread id = {threading.current_thread().ident}")
        time.sleep(1)


TaskThread = threading.Thread(target=run)
TaskThread.setDaemon(True)
TaskThread.start()

while True:
    time.sleep(3)
    stop_thread(TaskThread)
    TaskThread = threading.Thread(target=run)
    TaskThread.setDaemon(True)
    TaskThread.start()

运行结果如下图所示