Python中实现本地状态监听与任务中断切换的轮询问题

原代码的核心问题 #

• 锁使用不当：normal_task中在持有state_lock的情况下调用time.sleep(1)，导致锁被长时间占用，其他线程无法及时修改状态，状态切换的响应延迟严重。
• 主线程阻塞：主循环中的join()会强制等待任务线程结束后才继续检查状态，回到了“必须等任务结束才能响应状态变化”的原始问题，完全失去了并行性。
• 线程安全隐患：fall_detected变量的读写没有加锁，多线程环境下可能出现数据竞争，导致状态判断错误。
• 任务中断机制低效：依赖轮询状态变量实现任务切换，若任务中有更长时间的阻塞操作，无法及时响应状态变化。

正确实现思路 #

1. 使用threading.Event实现可中断的任务，相比轮询状态变量，事件的响应更及时、资源消耗更低。
2. 最小化锁的持有时间：只在读写共享状态变量时加锁，避免在锁内执行耗时操作（如sleep）。
3. 主线程负责状态监听与任务调度，不阻塞等待任务完成，而是通过事件和标志位控制任务线程的启停。
4. 统一管理共享状态的线程安全，所有对共享变量的读写操作都必须加锁。

修正后的代码 #

import random
import threading
import time

# 线程安全的状态管理
state_lock = threading.Lock()
current_state = 1  # 1: 正常状态, 2: 跌倒状态
# 用于中断正常任务的事件
normal_task_stop_event = threading.Event()

def normal_task():
    """可中断的正常任务"""
    task_duration = 20
    elapsed = 0
    normal_task_stop_event.clear()  # 重置中断事件
    while elapsed < task_duration and not normal_task_stop_event.is_set():
        print("Normal Task")
        time.sleep(1)
        elapsed += 1
        # 短时间检查状态，避免锁占用过久
        with state_lock:
            if current_state != 1:
                normal_task_stop_event.set()
    print("Normal Task exited")

def fall_task():
    """跌倒处理任务"""
    print("Executing Fall Detected")
    time.sleep(2)
    # 处理完成后恢复正常状态
    with state_lock:
        global current_state
        current_state = 1
    print("Fall handled")

def fall_detection_signal():
    """模拟跌倒检测信号触发"""
    while True:
        check_interval = random.uniform(5, 10)
        time.sleep(check_interval)
        print(f"Fall detection check completed after {check_interval:.2f}s")
        # 触发跌倒状态
        with state_lock:
            global current_state
            current_state = 2
        print("Fall Detected! Switching state to Fall Detected.")
        time.sleep(5)

def state_monitor():
    """状态监听与任务调度主线程"""
    normal_thread = None
    while True:
        with state_lock:
            current = current_state
        
        if current == 1:
            # 正常状态：确保正常任务在运行，若未运行则启动
            if not (normal_thread and normal_thread.is_alive()):
                normal_thread = threading.Thread(target=normal_task, daemon=True)
                normal_thread.start()
        elif current == 2:
            # 跌倒状态：中断正常任务，执行跌倒处理
            if normal_thread and normal_thread.is_alive():
                normal_task_stop_event.set()
                normal_thread.join()  # 等待正常任务退出
            fall_task()
        time.sleep(0.1)  # 降低轮询频率，减少资源消耗

# 启动跌倒检测模拟线程
fall_detection_thread = threading.Thread(target=fall_detection_signal, daemon=True)
fall_detection_thread.start()

# 启动状态监控与调度线程
state_monitor()

代码改进点说明 #

• threading.Event实现中断：normal_task_stop_event可以立即触发正常任务的退出，无需等待循环迭代完成，响应速度接近实时。
• 锁的优化：仅在读写current_state时加锁，锁的持有时间极短，不会阻塞其他线程的状态修改。
• 任务调度逻辑：state_monitor线程负责监控状态，自动启动/终止任务，主线程无需阻塞等待任务完成。
• 守护线程：将任务线程设为守护线程，避免程序退出时残留未结束的线程。

内容的提问来源于stack exchange，提问作者souren.p