如果您显式地调用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)，并且在我看来，处理退出的代码明显更简单。

在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,  *args, **kwargs):
        super(StoppableThread, self).__init__(*args, **kwargs)
        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 an 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 exception in the context of the
        thread represented by this instance.
        """
        _async_raise( self._get_my_tid(), exctype )

(根据Tomer Filiba的《Killable Threads》改编。关于PyThreadState_SetAsyncExc返回值的引用似乎来自旧版本的Python。)

正如文档中所指出的，这不是一个神奇的子弹，因为如果线程在Python解释器之外忙，它将不会捕获中断。

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

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

t是线程对象。

阅读python源代码(Modules/threadmodule.c和python /thread_pthread.h)，你可以看到Thread。Ident是一个pthread_t类型，所以你可以使用libpthread在python中做任何pthread可以做的事情。

另一种方法是使用signal.pthread_kill发送一个停止信号。

from signal import pthread_kill, SIGTSTP
from threading import Thread
from itertools import count
from time import sleep

def target():
    for num in count():
        print(num)
        sleep(1)

thread = Thread(target=target)
thread.start()
sleep(5)
pthread_kill(thread.ident, SIGTSTP)

结果

0
1
2
3
4

[14]+  Stopped

下面的方法可以用来杀死一个线程:

kill_threads = False

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

thread.start_new_thread(doSomething, ())

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

我通常使用它在程序出口终止所有线程。这可能不是终止线程的完美方法，但可能有帮助。

你不应该在没有与线程合作的情况下强行终止线程。

杀死一个线程消除了try/finally阻塞设置的任何保证，所以你可能会让锁锁定，文件打开等等。

唯一可以认为强制终止线程是一个好主意的情况是快速终止程序，但绝不是单个线程。