文章目录
1 threading.Event()方法2 子线程抛出异常，立刻停止
1 threading.Event()方法
一种常见的方法是使用标志位来通知线程应该停止。线程可以定期检查这个标志位，如果它被设置为停止，那么线程就结束其执行。下面是一个简单的例子：
import threading import time class MyThread(threading.Thread): def __init__(self): super(MyThread, self).__init__() self.stop_event = threading.Event() def run(self): # 清空设置 self.stop_event.clear() while not self.stop_event.is_set(): print("Thread is running...") time.sleep(1) def stop(self): self.stop_event.set() # 创建线程 thread = MyThread() thread.start() # 在某个时间点，停止线程 time.sleep(5) thread.stop()
需要注意的是，这只是一种优雅的停止线程的方法，它依赖于线程在
run方法中定期检查
stop_event。如果线程没有这样的检查，或者它正在执行一个无法被中断的阻塞操作（例如
IO操作），那么这种方法可能无法立即停止线程。
2 子线程抛出异常，立刻停止
在
Python中，使用
ctypes和
PyThreadState_SetAsyncExc函数来在子线程中异步抛出一个异常是一种相对底层的做法，它可以直接在子线程的上下文中触发一个异常。然而，这种做法需要谨慎使用，因为它可能会导致线程状态不稳定或未定义的行为，特别是如果线程没有正确地处理异常。
import threadingimport timeimport ctypesimport inspectdef do_some_task(): while True: time.sleep(1) print("子线程！1") time.sleep(1) print("子线程！2") time.sleep(1) print("子线程！3") time.sleep(1) print("子线程！4") time.sleep(1) print("子线程！5")def async_raise(thread_id, exctype): """ 通过C语言的库抛出异常 :param thread_id: :param exctype: :return: """ # 在子线程内部抛出一个异常结束线程 thread_id = ctypes.c_long(thread_id) if not inspect.isclass(exctype): exctype = type(exctype) res = ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, ctypes.py_object(exctype)) if res == 0: raise ValueError("线程id违法") elif res != 1: ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, None) raise SystemError("异常抛出失败")def stop_thread_now(thread): # 结束线程 async_raise(thread.ident, SystemExit)if __name__ == "__main__": # 可以在子线程任何时候随时结束子线程 sub_thread = threading.Thread(target=do_some_task,name="sub_thread") sub_thread.start() print(sub_thread.is_alive()) time.sleep(7) stop_thread_now(sub_thread) time.sleep(1) print(sub_thread.is_alive())