Windows环境下Python如何强制终止或join指定运行时长的线程

原有代码问题梳理 #

• 线程初始化错误：threading.Thread(target=func()) 写法会直接执行func函数而非传入函数引用，正确写法应为target=func
• 计时逻辑错误：time.perf_counter是计时函数，需要调用time.perf_counter()才能获取当前时间戳
• join方法使用错误：无参数的join()会一直阻塞直到线程结束，不会返回布尔值，根本无法进入后续超时判断逻辑

推荐实现方案（跨平台兼容Windows，无资源泄漏风险） #

推荐使用协作式终止方案，通过threading.Event作为停止信号，目标函数主动检测信号退出，该方案完全兼容Windows/Linux，不会出现强制杀线程导致的资源泄漏问题。

示例代码 #

import time
import threading

# 停止信号标志
stop_event = threading.Event()

def func():
    for i in range(0,7):
        # 每次循环先检测是否收到停止信号
        if stop_event.is_set():
            print("收到终止信号，提前退出")
            break
        print("not yet done")
        time.sleep(1)

if __name__ == "__main__":
    t1 = threading.Thread(target=func)
    t1.start()
    # 等待线程执行，最多等5秒
    t1.join(timeout=5)
    # 如果等待后线程还在运行，触发停止信号
    if t1.is_alive():
        stop_event.set()
        # 等待线程正常退出
        t1.join()
        print("已超时终止线程")
    print("执行结束")

特殊场景：无法修改目标函数时的强制终止方案 #

如果目标函数无法修改插入信号检测逻辑，可以通过ctypes调用Windows原生API强制终止线程，注意该方案存在资源泄漏风险，非必要不使用：

示例代码 #

import time
import threading
import ctypes

def func():
    for i in range(0,7):
        print("not yet done")
        time.sleep(1)

#  Windows API 终止线程封装
def terminate_thread(thread):
    if not thread.is_alive():
        return
    thread_id = thread.ident
    handle = ctypes.windll.kernel32.OpenThread(1, False, thread_id)
    ctypes.windll.kernel32.TerminateThread(handle, 0)
    ctypes.windll.kernel32.CloseHandle(handle)

if __name__ == "__main__":
    t1 = threading.Thread(target=func)
    t1.start()
    t1.join(timeout=5)
    if t1.is_alive():
        terminate_thread(t1)
        print("已强制终止超时线程")
    print("执行结束")

内容的提问来源于stack exchange，提问作者Crisóstomo Musa