参考:
Python进阶之路 - Timeout | 超时中断 - 知乎 (zhihu.com)
写的很棒。
这里只记录我摘取的封装的一个class:
#!/usr/bin/env python
# -*- coding: utf-8 -*-import ctypes
import threadingclass ThreadKillOver(RuntimeError):"""线程杀死后的反馈"""def __init__(self):RuntimeError.__init__(self, "ThreadKillOver")class ThreadKiller(threading.Thread):"""separate thread to kill TerminableThread"""def __init__(self, target_thread, exception_cls, repeat_sec=2.0):threading.Thread.__init__(self)self.target_thread = target_threadself.exception_cls = exception_clsself.repeat_sec = repeat_secself.daemon = Truedef run(self):"""loop raising exception incase it's caught hopefully this breaks us far out"""while self.target_thread.is_alive():#ctypes.pythonapi.PyThreadState_SetAsyncExc是Python的ctypes库中的一个函数,# 用于设置Python线程的异步异常。这个函数的第一个参数是一个C语言中的线程状态对象,# 第二个参数是要抛出的异常。这个函数允许你在Python的C API层面控制线程的异常处理,# 通常情况下,你不需要直接使用这个函数,而是让Python的异常处理机制自动处理。ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(self.target_thread.ident),ctypes.py_object(self.exception_cls))self.target_thread.join(self.repeat_sec)class TerminableThread(threading.Thread):"""a thread that can be stopped by forcing an exception in the execution context"""def terminate(self, exception_cls=ThreadKillOver, repeat_sec=1.0):"""一个是exception_cls,这个的含义是,线程结束时会raise哪一种异常;另一个是repeat_sec,这个的含义是,杀手每隔多长时间去确认一次线程有没有死亡。如果repeat_sec为1,则线程杀手会每秒检查一次,如果线程还没死,就再杀一次,知道确认死亡为止。"""if self.is_alive() is False:return Truekiller = ThreadKiller(self, exception_cls, repeat_sec=repeat_sec)killer.start()