在Python中，你不能直接杀死一个线程。

如果你不是真的需要Thread(!)，你可以做的，而不是使用threading包，是使用 多处理包。在这里，要终止一个进程，你可以简单地调用方法:

yourProcess.terminate()  # kill the process!

Python将终止您的进程(在Unix上通过SIGTERM信号，而在Windows上通过TerminateProcess()调用)。注意在使用队列或管道时使用它!(可能会破坏队列/管道中的数据)

注意，多处理。事件和多处理。信号量的工作方式与线程完全相同。事件和线程。信号量分别。事实上，前者是后者的克隆。

如果你真的需要使用一个线程，没有办法直接杀死它。但是，您可以使用“守护线程”。事实上，在Python中，线程可以被标记为守护进程:

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

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

注意，在调用start()方法之前，有必要将线程设置为守护进程!

当然，你可以，也应该在多进程中使用daemon。在这里，当主进程退出时，它将尝试终止其所有守护子进程。

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

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

kill_threads = False

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

thread.start_new_thread(doSomething, ())

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

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

在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解释器之外忙，它将不会捕获中断。

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

Python版本:3.8

使用守护线程来执行我们想要的，如果我们想要终止守护线程，我们只需要让父线程退出，然后系统就会终止父线程创建的守护线程。

还支持协程和协程函数。

def main():
    start_time = time.perf_counter()
    t1 = ExitThread(time.sleep, (10,), debug=False)
    t1.start()
    time.sleep(0.5)
    t1.exit()
    try:
        print(t1.result_future.result())
    except concurrent.futures.CancelledError:
        pass
    end_time = time.perf_counter()
    print(f"time cost {end_time - start_time:0.2f}")

下面是ExitThread源代码

import concurrent.futures
import threading
import typing
import asyncio


class _WorkItem(object):
    """ concurrent\futures\thread.py

    """

    def __init__(self, future, fn, args, kwargs, *, debug=None):
        self._debug = debug
        self.future = future
        self.fn = fn
        self.args = args
        self.kwargs = kwargs

    def run(self):
        if self._debug:
            print("ExitThread._WorkItem run")
        if not self.future.set_running_or_notify_cancel():
            return

        try:
            coroutine = None
            if asyncio.iscoroutinefunction(self.fn):
                coroutine = self.fn(*self.args, **self.kwargs)
            elif asyncio.iscoroutine(self.fn):
                coroutine = self.fn
            if coroutine is None:
                result = self.fn(*self.args, **self.kwargs)
            else:
                result = asyncio.run(coroutine)
            if self._debug:
                print("_WorkItem done")
        except BaseException as exc:
            self.future.set_exception(exc)
            # Break a reference cycle with the exception 'exc'
            self = None
        else:
            self.future.set_result(result)


class ExitThread:
    """ Like a stoppable thread

    Using coroutine for target then exit before running may cause RuntimeWarning.

    """

    def __init__(self, target: typing.Union[typing.Coroutine, typing.Callable] = None
                 , args=(), kwargs={}, *, daemon=None, debug=None):
        #
        self._debug = debug
        self._parent_thread = threading.Thread(target=self._parent_thread_run, name="ExitThread_parent_thread"
                                               , daemon=daemon)
        self._child_daemon_thread = None
        self.result_future = concurrent.futures.Future()
        self._workItem = _WorkItem(self.result_future, target, args, kwargs, debug=debug)
        self._parent_thread_exit_lock = threading.Lock()
        self._parent_thread_exit_lock.acquire()
        self._parent_thread_exit_lock_released = False  # When done it will be True
        self._started = False
        self._exited = False
        self.result_future.add_done_callback(self._release_parent_thread_exit_lock)

    def _parent_thread_run(self):
        self._child_daemon_thread = threading.Thread(target=self._child_daemon_thread_run
                                                     , name="ExitThread_child_daemon_thread"
                                                     , daemon=True)
        self._child_daemon_thread.start()
        # Block manager thread
        self._parent_thread_exit_lock.acquire()
        self._parent_thread_exit_lock.release()
        if self._debug:
            print("ExitThread._parent_thread_run exit")

    def _release_parent_thread_exit_lock(self, _future):
        if self._debug:
            print(f"ExitThread._release_parent_thread_exit_lock {self._parent_thread_exit_lock_released} {_future}")
        if not self._parent_thread_exit_lock_released:
            self._parent_thread_exit_lock_released = True
            self._parent_thread_exit_lock.release()

    def _child_daemon_thread_run(self):
        self._workItem.run()

    def start(self):
        if self._debug:
            print(f"ExitThread.start {self._started}")
        if not self._started:
            self._started = True
            self._parent_thread.start()

    def exit(self):
        if self._debug:
            print(f"ExitThread.exit exited: {self._exited} lock_released: {self._parent_thread_exit_lock_released}")
        if self._parent_thread_exit_lock_released:
            return
        if not self._exited:
            self._exited = True
            if not self.result_future.cancel():
                if self.result_future.running():
                    self.result_future.set_exception(concurrent.futures.CancelledError())

这是一个糟糕的答案，请看评论

以下是如何做到这一点:

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

这是基于thread2—可杀死线程ActiveState配方。

您需要调用PyThreadState_SetAsyncExc()，它只能通过ctypes模块使用。

这只在Python 2.7.3上进行了测试，但它很可能与最近的其他2一起工作。x版本。PyThreadState_SetAsyncExc()仍然存在于Python 3中以实现向后兼容性(但我还没有测试它)。

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")