import inspect
import ctypes
import time
import random
import traceback
from threading import Thread
import sys

# from utils.thread_killer import stop_thread


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):
    print(f'\nt -> {thread}')
    print(f'\nthread.ident -> {thread.ident}')
    _async_raise(thread.ident, SystemExit)


def func1():
    while True:
        try:
            print(f'func1\n')
            time.sleep(2)
        except SystemExit:
            sys.exit()
        except:
            print(traceback.format_exc())


def func2():
    while True:
        try:
            print(f'func2]\n')
            time.sleep(1.3)
        except SystemExit:
            sys.exit()
        except:
            print(traceback.format_exc())


def func3():
    try:
        func1()
    except SystemExit:
        sys.exit()
    except:
        print(traceback.format_exc())


class Test(object):
    def __init__(self):
        pass

    def engine(self):
        thread_pool = []
        t1 = Thread(target=func3, args=())
        t2 = Thread(target=func2, args=())
        thread_pool.append(t1)
        thread_pool.append(t2)
        for i in thread_pool:
            i.start()

        while True:
            try:
                time.sleep(3)
                # for i in thread_pool:
                #     print(f'線程{i}的狀態(tài){i.is_alive()}\n')
                #     print(f'線程{i}的名字{i.name}\n')
                #     print(f'線程的方法{i.ident}\n')
                #     time.sleep(1)

                for i in thread_pool:
                    print(type(i.name))
                    if i.name == 'Thread-1':
                        print(f'開始殺線程')
                        stop_thread(i)
                        time.sleep(3)
                        thread_pool.remove(i)
                        print(f'thread_pool -> {thread_pool}')
                        print(i.is_alive())
            except:
                pass


if __name__ == '__main__':
    test_case = Test()
    test_case.engine()

stop_thread 和 _async_raise的出處是[https://blog.csdn.net/Vinsuan1993/article/details/78158589]

要注意一定要在預(yù)備被殺的線程中攔截SystemExit錯誤，然后sys.exit()悯蝉，否則不能殺死該線程。
關(guān)鍵字：python 線程 殺 動態(tài)