强制退出运行python中的线程[duplicate]

在python和任何语言中,突然终止线程通常都是一种糟糕的模式。请考虑以下情况:

• 线程正在保存一个必须正确关闭的关键资源
• 该线程已经创建了几个其他线程,这些线程也必须被终止。

如果您负担得起(如果您管理自己的线程),处理这个问题的好方法是有一个exit_request标志,每个线程都会定期检查这个标志,看是否是时候让它退出了。

例如:

import threading

class StoppableThread(threading.Thread):
    """Thread class with a stop() method. The thread itself has to check
    regularly for the stopped() condition."""

    def __init__(self):
        super(StoppableThread, self).__init__()
        self._stop_event = threading.Event()

    def stop(self):
        self._stop_event.set()

    def stopped(self):
        return self._stop_event.is_set()

在这段代码中,您应该在希望线程退出时对其调用stop(),并使用join()等待线程正确退出。线程应定期检查停止标志。

但也有一些情况下,你真的需要杀死一个线程。一个例子是,当您包装一个外部库时,该库正忙着长时间的调用,并且您想中断它。

以下代码允许(有一些限制)在python线程中引发异常:

def _async_raise(tid, exctype):
    '''Raises an exception in the threads with id tid'''
    if not inspect.isclass(exctype):
        raise TypeError("Only types can be raised (not instances)")
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(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(ctypes.c_long(tid), None)
        raise SystemError("PyThreadState_SetAsyncExc failed")

class ThreadWithExc(threading.Thread):
    '''A thread class that supports raising exception in the thread from
       another thread.
    '''
    def _get_my_tid(self):
        """determines this (self's) thread id

        CAREFUL : this function is executed in the context of the caller
        thread, to get the identity of the thread represented by this
        instance.
        """
        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

        # TODO: in python 2.6, there's a simpler way to do : self.ident

        raise AssertionError("could not determine the thread's id")

    def raiseExc(self, exctype):
        """Raises the given exception type in the context of this thread.

        If the thread is busy in a system call (time.sleep(),
        socket.accept(), ...), the exception is simply ignored.

        If you are sure that your exception should terminate the thread,
        one way to ensure that it works is:

            t = ThreadWithExc( ... )
            ...
            t.raiseExc( SomeException )
            while t.isAlive():
                time.sleep( 0.1 )
                t.raiseExc( SomeException )

        If the exception is to be caught by the thread, you need a way to
        check that your thread has caught it.

        CAREFUL : this function is executed in the context of the
        caller thread, to raise an excpetion in the context of the
        thread represented by this instance.
        """
        _async_raise( self._get_my_tid(), exctype )

基于 Killable Threads 托默·菲利巴。关于 PyThreadState_SetAsyncExc 似乎来自 old version of Python )

如文档中所述,这不是一个神奇的项目符号,因为如果线程在python解释器之外很忙,它将不会捕捉到中断。

这段代码的一个好的使用模式是让线程捕获一个特定的异常并执行清理。这样,你可以中断一个任务,仍然有适当的清理。

没有官方的api可以做到这一点,不。

您需要使用平台api来终止线程,例如pthread_kill或terminatethread。您可以通过pythonwin或cTypes访问此类API。

请注意,这本质上是不安全的。它可能会导致无法收集的垃圾(来自成为垃圾的堆栈帧的局部变量),如果被杀死的线程在被杀死的时候具有gil,则可能导致死锁。

一 multiprocessing.Process 可以 p.terminate()

如果我想杀死一个线程,但不想使用flags/locks/signals/semaphores/events/whatever,我会将线程提升为完全成熟的进程。对于只使用几个线程的代码,开销并没有那么大。

例如,这可以方便地终止执行阻塞I/O的助手“线程”

转换很简单:在相关代码中全部替换 threading.Thread 具有 多处理过程 以及所有 queue.Queue 具有 multiprocessing.Queue 并添加所需的调用 P.终止() 你的父进程想要杀死它的子进程 p

Python doc

如果试图终止整个程序,可以将线程设置为“守护进程”。看见 Thread.daemon

这是基于 thread2 -- killable threads (Python recipe)

您需要调用Pythreadstate_setAsyncExc(),它仅通过cTypes可用。

这只在Python2.7.3上进行过测试,但它很可能适用于其他最近的2.x版本。

import ctypes

def terminate_thread(thread):
    """Terminates a python thread from another thread.

    :param thread: a threading.Thread instance
    """
    if not thread.isAlive():
        return

    exc = ctypes.py_object(SystemExit)
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(
        ctypes.c_long(thread.ident), exc)
    if res == 0:
        raise ValueError("nonexistent 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(thread.ident, None)
        raise SystemError("PyThreadState_SetAsyncExc failed")

你不应该不配合就强行杀死一根线。

终止线程将删除try/finally块设置的任何保证,这样您可能会将锁锁定、文件打开等。

你唯一能说强制杀死线程是个好主意的时候,就是快速杀死一个程序,但不要只杀死一个线程。

正如其他人所提到的,标准是设置一个停止标志。对于轻量级的东西(没有线程的子类化,没有全局变量),lambda回调是一个选项。(注意括号 if stop() )

import threading
import time

def do_work(id, stop):
    print("I am thread", id)
    while True:
        print("I am thread {} doing something".format(id))
        if stop():
            print("  Exiting loop.")
            break
    print("Thread {}, signing off".format(id))


def main():
    stop_threads = False
    workers = []
    for id in range(0,3):
        tmp = threading.Thread(target=do_work, args=(id, lambda: stop_threads))
        workers.append(tmp)
        tmp.start()
    time.sleep(3)
    print('main: done sleeping; time to stop the threads.')
    stop_threads = True
    for worker in workers:
        worker.join()
    print('Finis.')

if __name__ == '__main__':
    main()

替代 print() 用一个 pr() 总是刷新的函数( sys.stdout.flush() )可以提高外壳输出的精度。

(仅在windows/eclipse/python3.3上测试)

在python中,不能直接杀死线程。

如果你真的不需要线程(!),而不是使用 threading package ,即使用 multiprocessing package . 在这里,要终止进程,只需调用方法:

yourProcess.terminate()  # kill the process!

python将终止您的进程(在unix上通过sigterm信号,而在windows上通过 TerminateProcess() 打电话)使用队列或管道时请注意使用!(它可能损坏队列/管道中的数据)

请注意 multiprocessing.Event 以及 multiprocessing.Semaphore 工作方式完全相同 threading.Event 以及 threading.Semaphore 分别是。事实上,第一批是Latters的克隆。

如果你真的需要使用线程,没有办法直接杀死它。但是,您可以做的是使用 “守护进程线程” . 实际上,在python中,线程可以标记为 守护进程 :

yourThread.daemon = True  # set the Thread as a "daemon thread"

当没有活动的非守护进程线程时,主程序将退出。换句话说,当主线程(当然是非守护进程线程)完成其操作时,即使仍有一些守护进程线程在工作,程序也将退出。

注意,有必要将线程设置为 daemon 之前 start() 方法被调用!

你当然可以也应该 守护进程 即使有 multiprocessing . 在这里,当主进程退出时,它会尝试终止其所有守护子进程。

最后,请注意 sys.exit() 和 os.kill() 不是选择。

可以通过在将退出线程的线程中安装跟踪来终止线程。一个可能的实现见附件链接。

Kill a thread in Python

如果你不杀线就更好了。 一种方法可以是在线程周期中引入一个“try”块,并在希望停止线程时抛出一个异常(例如break/return/……这就阻止了你的。 我已经在我的应用程序中使用过这个,它可以工作…

完全有可能实现 Thread.stop 方法,如以下示例代码所示:

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 班级。

from ctypes import *
pthread = cdll.LoadLibrary("libpthread-2.15.so")
pthread.pthread_cancel(c_ulong(t.ident))

T 是你的 Thread 对象。

阅读python源代码( Modules/threadmodule.c 和 Python/thread_pthread.h )你可以看到 Thread.ident 是一个 pthread_t 打字,所以你可以做任何事 pthread 可以在python中使用 libpthread 。

以下解决方法可用于终止线程:

kill_threads = False

def doSomething():
    global kill_threads
    while True:
        if kill_threads:
            thread.exit()
        ......
        ......

thread.start_new_thread(doSomething, ())

这甚至可以用于从主线程终止其代码在另一个模块中写入的线程。我们可以在该模块中声明一个全局变量,并使用它终止在该模块中生成的线程。

我通常使用这个来终止程序出口处的所有线程。这可能不是终止线程的完美方法,但可能会有所帮助。

我想补充一点,如果你在 threading lib Python ,建议避免使用“恶魔”线程,当您不希望线程突然结束时,使用paolo rovelli标志 mentioned 。

官方文件:

守护进程线程在关闭时突然停止。它们的资源(如打开的文件、数据库事务等)可能无法正确释放。如果希望线程正常停止,请使它们非守护进程,并使用适当的信令机制(如事件)。

我认为创建守护线程取决于您的应用程序,但通常(在我看来)最好避免杀死它们或使它们成为守护线程。在多处理中,您可以使用 is_alive() 检查进程状态并“终止”以完成它们(也可以避免gil问题)。但有时,当你在windows中执行代码时,你会发现更多的问题。

永远记住,如果有“活线程”,python解释器将运行等待它们。(因为这个守护可以帮助你如果不要紧的话突然结束)。

我来晚了,但我一直在努力 a similar question 下面的代码似乎可以很好地解决这个问题,并让我在守护子线程退出时执行一些基本的线程状态检查和清理:

import threading
import time
import atexit

def do_work():

  i = 0
  @atexit.register
  def goodbye():
    print ("'CLEANLY' kill sub-thread with value: %s [THREAD: %s]" %
           (i, threading.currentThread().ident))

  while True:
    print i
    i += 1
    time.sleep(1)

t = threading.Thread(target=do_work)
t.daemon = True
t.start()

def after_timeout():
  print "KILL MAIN THREAD: %s" % threading.currentThread().ident
  raise SystemExit

threading.Timer(2, after_timeout).start()

产量:

0
1
KILL MAIN THREAD: 140013208254208
'CLEANLY' kill sub-thread with value: 2 [THREAD: 140013674317568]

如果您明确地调用 time.sleep() 作为线程的一部分(比如轮询一些外部服务),Phillipe方法的一个改进是在 event 的 wait() 方法无论你在哪里 sleep()

例如:

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() 而不是 睡眠() 定期检查事件是,您可以在较长的睡眠时间间隔内编程,线程几乎立即停止(否则 小精灵 在我看来,处理exit的代码要简单得多。

有一个图书馆是为这个目的而建的, stopit . 尽管这里列出的一些相同的警告仍然适用,但是至少这个库提供了一种常规的、可重复的技术来实现所述的目标。

这似乎适用于windows 7上的pywin32

my_thread = threading.Thread()
my_thread.start()
my_thread._Thread__stop()

这是个错误的答案,请看评论

以下是操作方法:

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() 方法,终止套接字,然后运行 thread._Thread__stop() 在你的内心 {) 。

使用setdaemon(true)启动子线程。

def bootstrap(_filename):
    mb = ModelBootstrap(filename=_filename) # Has many Daemon threads. All get stopped automatically when main thread is stopped.

t = threading.Thread(target=bootstrap,args=('models.conf',))
t.setDaemon(False)

while True:
    t.start()
    time.sleep(10) # I am just allowing the sub-thread to run for 10 sec. You can listen on an event to stop execution.
    print('Thread stopped')
    break

虽然它很老, this 对某些人来说可能是个方便的解决方案:

扩展线程模块功能的小模块-- 允许一个线程在另一个线程的上下文中引发异常 线程。通过提高 SystemExit ,最终可以杀死python线程。

import threading
import ctypes     

def _async_raise(tid, excobj):
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(excobj))
    if res == 0:
        raise ValueError("nonexistent 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 Thread(threading.Thread):
    def raise_exc(self, excobj):
        assert self.isAlive(), "thread must be started"
        for tid, tobj in threading._active.items():
            if tobj is self:
                _async_raise(tid, excobj)
                return

        # the thread was alive when we entered the loop, but was not found 
        # in the dict, hence it must have been already terminated. should we raise
        # an exception here? silently ignore?

    def terminate(self):
        # must raise the SystemExit type, instead of a SystemExit() instance
        # due to a bug in PyThreadState_SetAsyncExc
        self.raise_exc(SystemExit)

因此,它允许一个“线程在另一个线程的上下文中引发异常”,这样,终止的线程就可以在不定期检查中止标志的情况下处理终止。

然而,根据其 original source ,此代码有一些问题。

• 只有在执行python字节码时才会引发异常。如果线程调用本机/内置的阻塞函数,则 只有当执行返回到python时才会引发异常 代码。
  • 如果内置函数内部调用pyerr_clear(),也会出现问题,这将有效地取消挂起的异常。 你可以再举一次。
• 只能安全地引发异常类型。异常实例可能导致意外行为,因此受到限制。
  • 例如:t1.raise_exc(typeerror)而不是t1.raise_exc(typeerror(“blah”)。
  • 嗯,这是个虫子,我把它当作虫子报告了。更多信息, http://mail.python.org/pipermail/python-dev/2006-August/068158.html
• 我要求在内置线程模块中公开这个函数,但是由于cTypes已经成为一个标准库(从2.5开始),所以
  功能不太可能是实现不可知的,它可以保留
  未暴露的

Pieter Hintjens是 ØMQ -project——也就是说,使用mq并避免同步原语(如锁、互斥锁、事件等)是编写多线程程序的最明智和最安全的方法:

http://zguide.zeromq.org/py:all#Multithreading-with-ZeroMQ

这包括告诉子线程它应该取消它的工作。这可以通过为线程配置mq套接字并在该套接字上轮询一条消息来完成,该消息表示该线程应该取消。

该链接还提供了一个使用mq的多线程python代码的示例。

您可以在进程中执行命令,然后使用进程id终止它。 我需要在两个线程之间同步其中一个线程本身不返回。

processIds = []

def executeRecord(command):
    print(command)

    process = subprocess.Popen(command, stdout=subprocess.PIPE)
    processIds.append(process.pid)
    print(processIds[0])

    #Command that doesn't return by itself
    process.stdout.read().decode("utf-8")
    return;


def recordThread(command, timeOut):

    thread = Thread(target=executeRecord, args=(command,))
    thread.start()
    thread.join(timeOut)

    os.kill(processIds.pop(), signal.SIGINT)

    return;

如果您真的需要终止子任务的能力,请使用另一个实现。 multiprocessing 和 gevent 两者都支持滥杀一根“线”。

python的线程不支持取消。甚至不要尝试。您的代码很可能会死锁、损坏或泄漏内存,或具有其他意外的“有趣”难以调试的效果,这种效果很少发生,而且不确定。