虽然它相当古老，但对于某些人来说，这可能是一个方便的解决方案：
一个扩展线程模块功能的小模块——允许一个线程在另一个线程的上下文中引发异常。提高 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)

因此，它允许“一个线程在另一个线程的上下文中引发异常”，这样，终止的threa

仅仅是基于@scb的想法（这正是我所需要的）创建一个具有自定义函数的killablethread子类：

from threading import Thread, Event

class KillableThread(Thread):
    def __init__(self, sleep_interval=1, target=None, name=None, args=(), kwargs={}):
        super().__init__(None, target, name, args, kwargs)
        self._kill = Event()
        self._interval = sleep_interval
        print(self._target)

    def run(self):
        while True:
            # Call custom function with arguments
            self._target(*self._args)

            # 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()

if __name__ == '__main__':

    def print_msg(msg):
        print(msg)

    t = KillableThread(10, print_msg, args=("hello world"))
    t.start()
    time.sleep(6)
    print("About to kill thread")
    t.kill()

自然地，就像@sbc一样，线程不必等待运行新循环就可以停止。在本例中，您将看到“killing thread”消息打印在“about to kill thread”之后，而不是再等待4秒钟线程完成（因为我们已经睡了6秒钟）。
killablethread构造函数中的第二个参数是自定义函数（此处打印消息）。args参数是在此处调用函数（（“hello world”）时将使用的参数。

下面是另一种方法，但是使用非常干净和简单的代码，它将在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/

有一个为此目的而建的图书馆，停止吧。尽管本文列出的一些注意事项仍然适用，但至少该库提供了一种用于实现所述目标的常规、可重复的技术。

我想补充的一点是，如果您阅读线程库python中的官方文档，建议避免使用“恶魔”线程，因为您不希望线程突然结束，并带有paolo rovelli提到的标志。
从正式文件：
守护进程线程在关闭时突然停止。他们的资源（如打开的文件、数据库事务等）可能无法正常释放。如果希望线程正常停止，请将其设为非守护线程，并使用适当的信号机制，如事件。
我认为创建daemonic线程取决于您的应用程序，但通常（在我看来）最好避免杀死它们或使它们成为daemonic线程。在多处理中，您可以使用 is_alive() 检查流程状态并“终止”完成它们（也可以避免gil问题）。但有时，当您在windows中执行代码时，您会发现更多问题。
请始终记住，如果您有“活动线程”，python解释器将为它们运行(因为这个恶魔可以帮你，如果不要紧的话（突然结束）。

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

这是你的 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, ())

这甚至可以用于从主线程终止线程（其代码写在另一个模块中）。我们可以在该模块中声明一个全局变量，并使用它终止在该模块中生成的线程。
我通常使用它来终止程序出口处的所有线程。这可能不是终止线程的完美方式，但可能会有所帮助。

我玩这个游戏已经晚了很久，但我一直在努力解决一个类似的问题，下面的内容似乎可以完美地解决这个问题，并让我在守护子线程退出时执行一些基本的线程状态检查和清理：

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]

最好不要断线。一种方法是在线程的循环中引入一个“try”块，并在想要停止线程时抛出一个异常（例如break/return/…）。。。这将停止您的操作（for/while/…）。我已经在我的应用程序中使用了它，它可以工作。。。

实施一个 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 班级。

在python和任何语言中，突然终止线程通常都是不好的模式。请考虑以下情况：
线程持有一个必须正确关闭的关键资源
该线程还创建了其他几个必须终止的线程。
如果您负担得起（如果您正在管理自己的线程），处理这个问题的好方法是使用一个exit_请求标志，每个线程定期检查该标志，以查看是否是退出的时间。
例如：

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() . 线程应定期检查停止标志。
但是，在某些情况下，确实需要终止线程。例如，当您正在 Package 一个因长时间呼叫而忙碌的外部库时，您想中断它。
以下代码允许（有一些限制）在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解释器之外忙碌，它将无法捕获中断。
这段代码的一个好的使用模式是让线程捕获特定的异常并执行清理。这样，您就可以中断一个任务，并且仍然可以进行适当的清理。

您可以通过将跟踪安装到将退出线程的线程中来终止线程。有关一种可能的实现方式，请参见所附链接。
在python中杀死线程

如果您明确地呼叫 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() 而不是 sleep() 如果您可以在更长的睡眠时间间隔内编程，那么线程几乎会立即停止（如果您不这样做的话） sleep() （ing）在我看来，处理退出的代码要简单得多。

在不与线程合作的情况下，决不能强行杀死线程。
终止一个线程将删除try/finally阻塞设置的任何保证，这样您就可以保持锁锁定、文件打开等状态。
你唯一可以说强制杀死线程是个好主意的时候就是快速杀死一个程序，但绝不是单个线程。

在python中，不能直接终止线程。
如果您真的不需要线程（！），您可以做的是使用多处理包，而不是使用线程包。在这里，要终止进程，只需调用以下方法：

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() 方法被调用！
当然，你可以也应该使用 daemon 即使 multiprocessing . 在这里，当主进程退出时，它会尝试终止其所有daemonic子进程。
最后，请注意 sys.exit() 及 os.kill() 这不是选择。