如何用wait替代Python中的sleep实现线程即时中断？

方案一：使用threading.Event改造多线程代码 #

如果你的代码基于多线程，threading.Event的wait()方法完美替代time.sleep()，既能实现定时休眠，又能在需要时即时中断线程。

改造示例 #

假设你原来的线程代码是这样的：

import threading
import time

def worker():
    while True:
        print("执行任务...")
        time.sleep(10)  # 此处休眠时无法被即时中断
        print("休眠结束，继续任务")

t = threading.Thread(target=worker)
t.start()

改成用Event实现可中断休眠的版本：

import threading

def worker(stop_event):
    while not stop_event.is_set():
        print("执行任务...")
        # 用wait替代sleep，超时10秒，同时监听中断信号
        stop_event.wait(10)
        # 检查是否收到中断信号
        if stop_event.is_set():
            print("收到中断，退出线程")
            break
        print("休眠结束，继续任务")

# 创建一个用于中断的Event对象
stop_event = threading.Event()
# 启动线程并传入Event
t = threading.Thread(target=worker, args=(stop_event,))
t.start()

# 示例：3秒后触发中断
import time
time.sleep(3)
stop_event.set()
t.join()

原理说明 #

• stop_event.wait(10)会阻塞线程最多10秒，如果期间stop_event.set()被调用，会立刻返回，不会等到超时。
• 通过循环检查stop_event.is_set()，线程可以在收到中断信号后及时退出，解决了time.sleep()无法被即时中断的问题。

方案二：使用asyncio实现异步协程的可中断休眠 #

如果你的代码适合用异步模型，asyncio提供了可取消的休眠机制，无需线程也能实现即时中断。

改造示例 #

import asyncio

async def worker():
    try:
        while True:
            print("执行异步任务...")
            await asyncio.sleep(10)  # 异步休眠，支持取消
            print("休眠结束，继续任务")
    except asyncio.CancelledError:
        # 捕获取消异常，做退出前的清理
        print("收到取消信号，退出协程")

async def main():
    # 创建异步任务
    task = asyncio.create_task(worker())
    # 3秒后取消任务
    await asyncio.sleep(3)
    task.cancel()
    # 等待任务处理完取消异常
    await task

# 运行异步主程序
asyncio.run(main())

原理说明 #

• asyncio.sleep()是异步非阻塞的，当调用task.cancel()时，会向协程抛出CancelledError异常。
• 协程捕获该异常后即可退出，实现“即时中断”的效果（注意：asyncio是协作式调度，取消信号会在协程的下一个await点生效）。

方案选择 #

• 如果你的代码原本是多线程架构，优先用threading.Event方案，改动最小。
• 如果是新写的IO密集型代码，推荐用asyncio方案，性能更高且无需线程切换开销。

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