Python如何从其他函数内部终止运行中的threading线程函数

Python 标准库的 threading.Thread 没有提供原生的强制终止线程的接口，最安全、最常用的实现方式是通过自定义退出标志位的方式实现： #

完整可运行修改后代码 #

from threading import *
from time import *
import keyboard

# 定义线程退出标志位，初始为False
exit_flag = False

def ExampleFunction():
    # 循环条件改为判断退出标志位
    while not exit_flag:
        print('The Example function is running.')  # 循环内执行相关操作
        sleep(1)
    # 可以在这里加线程退出前的资源清理逻辑
    print("ExampleFunction 已正常退出")

def ExitFunction():
    global exit_flag
    print('ExitFunction was called.')
    keyboard.wait('f8')  # 等待f8按键被按下
    # 修改退出标志位，通知目标线程退出
    exit_flag = True

if __name__ == '__main__':
    Thread(target=ExampleFunction).start()
    Thread(target=ExitFunction).start()

实现说明 #

• 核心逻辑是用一个布尔变量作为两个线程的共享通信标志，不需要加锁，因为Python中布尔值的赋值是原子操作
• 如果你的业务逻辑中循环内有超过1秒的阻塞操作，建议把长阻塞拆分为短时间轮询，同时每次轮询都判断退出标志位，避免线程退出延迟
• 不推荐使用操作系统层面强制kill线程的方案，该方案容易导致线程持有的锁、打开的文件等资源没有正常释放，引发内存泄漏、死锁等不可预期的问题

内容的提问来源于stack exchange，提问作者Simon Bretschneider