是否有可能终止一个正在运行的线程而不设置/检查任何标志/信号/等等?
当前回答
我想补充的一件事是,如果你阅读threading lib Python的官方文档,建议避免使用“恶魔”线程,当你不希望线程突然结束时,带有Paolo Rovelli提到的标志。
来自官方文件:
守护进程线程在关机时突然停止。它们的资源(如打开的文件、数据库事务等)可能无法正确释放。如果您希望线程优雅地停止,请将它们设置为非守护进程,并使用适当的信号机制(如Event)。
我认为创建守护线程取决于您的应用程序,但通常(在我看来)最好避免杀死它们或使它们成为守护线程。在多处理中,您可以使用is_alive()来检查进程状态,并使用“terminate”来完成它们(也可以避免GIL问题)。但有时,当你在Windows中执行代码时,你会发现更多的问题。
并且永远记住,如果你有“活动线程”,Python解释器将运行等待它们。(因为这个守护程序可以帮助你如果不重要的事情突然结束)。
其他回答
这里还有另一种方法,但代码非常干净和简单,适用于2021年的Python 3.7:
import ctypes
def kill_thread(thread):
"""
thread: a threading.Thread object
"""
thread_id = thread.ident
res = ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, ctypes.py_object(SystemExit))
if res > 1:
ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, 0)
print('Exception raise failure')
改编自:https://www.geeksforgeeks.org/python-different-ways-to-kill-a-thread/
这是一个糟糕的答案,请看评论
以下是如何做到这一点:
from threading import *
...
for thread in enumerate():
if thread.isAlive():
try:
thread._Thread__stop()
except:
print(str(thread.getName()) + ' could not be terminated'))
给它几秒钟,然后你的线程应该停止。还要检查thread._Thread__delete()方法。
为了方便,我建议使用thread.quit()方法。例如,如果你的线程中有一个套接字,我建议在你的套接字句柄类中创建一个quit()方法,终止套接字,然后在你的quit()中运行一个thread. _thread__stop()。
如果您显式地调用time.sleep()作为线程的一部分(比如轮询一些外部服务),对Phillipe方法的改进是在sleep()的任何地方使用事件的wait()方法中的超时。
例如:
import threading
class KillableThread(threading.Thread):
def __init__(self, sleep_interval=1):
super().__init__()
self._kill = threading.Event()
self._interval = sleep_interval
def run(self):
while True:
print("Do Something")
# If no kill signal is set, sleep for the interval,
# If kill signal comes in while sleeping, immediately
# wake up and handle
is_killed = self._kill.wait(self._interval)
if is_killed:
break
print("Killing Thread")
def kill(self):
self._kill.set()
然后运行它
t = KillableThread(sleep_interval=5)
t.start()
# Every 5 seconds it prints:
#: Do Something
t.kill()
#: Killing Thread
使用wait()而不是sleep()ing和定期检查事件的优点是,您可以在更长的睡眠间隔中编程,线程几乎立即停止(否则您将处于sleep()ing),并且在我看来,处理退出的代码明显更简单。
实现一个线程是绝对可能的。方法,如下例代码所示:
import sys
import threading
import time
class StopThread(StopIteration):
pass
threading.SystemExit = SystemExit, StopThread
class Thread2(threading.Thread):
def stop(self):
self.__stop = True
def _bootstrap(self):
if threading._trace_hook is not None:
raise ValueError('Cannot run thread with tracing!')
self.__stop = False
sys.settrace(self.__trace)
super()._bootstrap()
def __trace(self, frame, event, arg):
if self.__stop:
raise StopThread()
return self.__trace
class Thread3(threading.Thread):
def _bootstrap(self, stop_thread=False):
def stop():
nonlocal stop_thread
stop_thread = True
self.stop = stop
def tracer(*_):
if stop_thread:
raise StopThread()
return tracer
sys.settrace(tracer)
super()._bootstrap()
###############################################################################
def main():
test1 = Thread2(target=printer)
test1.start()
time.sleep(1)
test1.stop()
test1.join()
test2 = Thread2(target=speed_test)
test2.start()
time.sleep(1)
test2.stop()
test2.join()
test3 = Thread3(target=speed_test)
test3.start()
time.sleep(1)
test3.stop()
test3.join()
def printer():
while True:
print(time.time() % 1)
time.sleep(0.1)
def speed_test(count=0):
try:
while True:
count += 1
except StopThread:
print('Count =', count)
if __name__ == '__main__':
main()
Thread3类运行代码的速度似乎比Thread2类快大约33%。
如果您试图终止整个程序,您可以将线程设置为“守护进程”。看到 Thread.daemon
推荐文章
- 证书验证失败:无法获得本地颁发者证书
- 当使用pip3安装包时,“Python中的ssl模块不可用”
- 无法切换Python与pyenv
- Python if not == vs if !=
- 如何从scikit-learn决策树中提取决策规则?
- 为什么在Mac OS X v10.9 (Mavericks)的终端中apt-get功能不起作用?
- 将旋转的xtick标签与各自的xtick对齐
- 为什么元组可以包含可变项?
- 如何合并字典的字典?
- 如何创建类属性?
- 不区分大小写的“in”
- 在Python中获取迭代器中的元素个数
- 自动化invokerrequired代码模式
- 解析日期字符串并更改格式
- 使用try和。Python中的if