另一种方法是使用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

最简单的方法是:

from threading import Thread
from time import sleep

def do_something():
    global thread_work
    while thread_work:
        print('doing something')
        sleep(5)
    print('Thread stopped')

thread_work = True
Thread(target=do_something).start()
sleep(5)
thread_work = False

这里还有另一种方法，但代码非常干净和简单，适用于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/

正如其他人所提到的，规范是设置一个停止标志。对于一些轻量级的东西(没有Thread的子类化，没有全局变量)，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())可以提高shell输出的精度。

(仅在Windows/Eclipse/Python3.3上测试)

多处理。进程可以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

实现一个线程是绝对可能的。方法，如下例代码所示:

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类快大约33%。