作者Lucemia (生の直感、死の予感)
看板Python
標題[心得] Haltable Thread
時間Sun Sep 16 00:14:38 2007
python 內建的thread 沒有提供終止執行功能
這個網路上看到版本的code 有提供 haltable thread 的一個實作方式 :)
from threading import Thread
def _async_raise(tid, exctype):
"""raises the exception, performs cleanup if needed"""
if not inspect.isclass(exctype):
raise TypeError("Only types can be raised (not instances)")
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, 0)
raise SystemError("PyThreadState_SetAsyncExc failed")
class yourthread(Thread):
def __init__ (self,imgurl):
Thread.__init__(self)
## your init here
def _get_my_tid(self):
"""determines this (self's) thread id"""
if not self.isAlive():
raise threading.ThreadError("the thread is not active")
# do we have it cached?
if hasattr(self, "_thread_id"):
return self._thread_id
# no, look for it in the _active dict
for tid, tobj in threading._active.items():
if tobj is self:
self._thread_id = tid
return tid
raise AssertionError("could not determine the thread's id")
def terminate(self):
"""raises SystemExit in the context of the given thread, which should
cause the thread to exit silently (unless caught)"""
_async_raise(self._get_my_tid(), SystemExit)
def run(self):
pass ## your code here
--
※ 發信站: 批踢踢實業坊(ptt.cc)
◆ From: 140.110.216.37
※ 編輯: Lucemia 來自: 140.110.216.38 (09/16 00:17)