如何在Python中中断线程？30秒后中断线程是否可行？

好问题！在Python的threading模块里，并没有直接强制中断线程的官方方法——这是因为Python的线程是协作式的，强制中断可能会导致资源泄漏（比如未释放的锁、打开的文件句柄），所以官方更推荐让线程主动检测退出信号来停止。不过我们可以通过优雅的方式实现你要的“30秒后中断线程”的需求，下面给你两种常用方案：

方案1：用threading.Event发送停止信号 #

threading.Event是线程安全的信号工具，主线程可以通过它给子线程发送“停止指令”，子线程只需要定期检查这个信号即可。

修改你的示例代码如下：

import threading
import time

def waiting(stop_event):
    print("thread is started")
    # 把长sleep拆分成多次短sleep，每次检查停止信号
    for _ in range(60):
        if stop_event.is_set():
            print("thread is interrupted")
            return
        time.sleep(1)
    print("thread is finished")

# 初始化一个事件对象
stop_event = threading.Event()
thread = threading.Thread(target=waiting, args=(stop_event,), name="thread_1")
thread.start()

# 30秒后触发停止信号
time.sleep(30)
stop_event.set()

这里把原来的time.sleep(60)拆成了60次1秒的休眠，这样子线程每秒都有机会检查是否收到停止信号。当主线程调用stop_event.set()后，子线程检测到信号就会主动退出。

方案2：自定义可停止的线程类 #

如果你需要更封装的实现，可以继承threading.Thread，给线程添加停止标志位和停止方法：

import threading
import time

class StoppableThread(threading.Thread):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._stop_flag = False

    def stop(self):
        # 设置停止标志（线程安全，因为布尔赋值是原子操作）
        self._stop_flag = True

    def run(self):
        print("thread is started")
        for _ in range(60):
            if self._stop_flag:
                print("thread is interrupted")
                return
            time.sleep(1)
        print("thread is finished")

# 创建自定义线程实例并启动
thread = StoppableThread(name="thread_1")
thread.start()

# 30秒后调用stop方法发送停止信号
time.sleep(30)
thread.stop()

这个方案把停止逻辑封装在线程类内部，代码结构更清晰。

补充说明：为什么不能强制中断？ #

Python的threading.Thread没有提供类似terminate()的强制终止方法，因为强制终止线程可能会让线程处于不一致的状态——比如线程正在持有锁时被中断，会导致其他线程永远无法获取锁，引发死锁；或者线程正在写入文件时被中断，导致文件损坏。所以协作式退出是Python中停止线程的最佳实践。

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