多处理。进程可以p.terminate()

如果我想杀死一个线程，但不想使用标志/锁/信号/信号量/事件/任何东西，我就把线程提升到完整的进程。对于只使用几个线程的代码，开销并没有那么糟糕。

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

转换很简单:在相关代码中替换所有线程。多线程线程。进程和所有队列。多处理队列。排队并将p.t terminate()所需的调用添加到想要杀死子进程p的父进程中

关于多处理，请参阅Python文档。

例子:

import multiprocessing
proc = multiprocessing.Process(target=your_proc_function, args=())
proc.start()
# Terminate the process
proc.terminate()  # sends a SIGTERM

Pieter Hintjens (ØMQ-project的创始人之一)表示，使用ØMQ并避免使用锁、互斥对象、事件等同步原语，是编写多线程程序最明智、最安全的方法:

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

这包括告诉子线程，它应该取消它的工作。这可以通过为线程配备ØMQ-socket并轮询该套接字以获得一条表示它应该取消的消息来实现。

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

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

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

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

只是建立在@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一样，线程不会等待运行一个新的循环来停止。在这个例子中，你会看到“kill Thread”消息紧跟在“About to kill Thread”之后，而不是等待4秒钟线程完成(因为我们已经睡了6秒了)。

KillableThread构造函数中的第二个参数是您的自定义函数(print_msg)。Args参数是在调用函数(("hello world"))时使用的参数。

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

以下是如何做到这一点:

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