如何在threading.Event触发时中途退出Python函数？

要实现在event触发时立即终止capture函数，核心思路是把不可中断的time.sleep()替换成threading.Event.wait()——这个方法会在超时（模拟原sleep时长）或event被设置时立即返回，完美解决中途退出的问题。

步骤1：修复原线程代码的bug #

原代码存在两处语法错误：

• threading.thread需改为大写的threading.Thread
• th1.start需要加括号调用：th1.start()

步骤2：改造capture函数 #

把每个time.sleep(t)替换成if event.wait(t): return——如果wait()返回True，说明event被触发，直接终止函数；如果返回False，说明超时，继续执行下一个任务。

改造后的完整代码示例：

import threading
import time

event = threading.Event()
# 模拟传感器状态，可根据实际逻辑修改
sensor = False

def eventSetter():
    global sensor
    while True:
        # 模拟传感器状态变化：3秒后触发，保持2秒后恢复
        time.sleep(3)
        sensor = True
        event.set()
        time.sleep(2)
        sensor = False
        event.clear()

def capture():
    # 第一个任务：原sleep 2秒，替换为可中断等待
    if event.wait(2):
        print("capture被中途终止（第一个任务执行中）")
        return
    print("第一个任务完成")
    
    # 第二个任务：原sleep 1秒
    if event.wait(1):
        print("capture被中途终止（第二个任务执行中）")
        return
    print("第二个任务完成")
    
    # 第三个任务：原sleep 2秒
    if event.wait(2):
        print("capture被中途终止（第三个任务执行中）")
        return
    print("所有任务完成")

# 启动eventSetter线程（设为守护线程，主程序退出时自动终止）
th1 = threading.Thread(target=eventSetter)
th1.daemon = True
th1.start()

# 测试capture函数
print("开始执行capture")
capture()
print("capture执行结束")

进一步优化：封装可中断sleep函数 #

如果有大量耗时任务，可以封装工具函数简化代码：

def interruptible_sleep(event, timeout):
    return event.wait(timeout)

def capture():
    if interruptible_sleep(event, 2):
        print("capture被中途终止（第一个任务执行中）")
        return
    print("第一个任务完成")
    
    if interruptible_sleep(event, 1):
        print("capture被中途终止（第二个任务执行中）")
        return
    print("第二个任务完成")
    
    if interruptible_sleep(event, 2):
        print("capture被中途终止（第三个任务执行中）")
        return
    print("所有任务完成")

原理说明 #

event.wait(timeout)的关键行为：

• 当event被设置时，立即返回True，无视timeout是否到期
• 当timeout到期但event未被设置时，返回False
  这就实现了在任何耗时任务的执行过程中，只要event触发，就能立刻终止capture函数，无需等待当前任务完成。

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