如何在指定时间后终止线程并保持主线程响应性

要实现指定时间后终止子线程，同时让处理用户输入的主线程保持活跃，核心思路是用线程安全的终止信号让子线程主动退出循环，结合定时器触发终止信号。Python没有直接强制终止线程的安全方法，让线程自行检测退出信号是最优方案。

具体实现步骤 #

• 使用threading.Event作为线程终止的信号，它能在多线程环境下安全传递"终止"指令。
• 修改子线程的循环逻辑，每次循环时检查终止信号，若信号触发则退出循环。
• 当用户输入命令4并指定时间后，创建新的子线程和终止信号，同时启动定时器，到指定时间后触发终止信号。
• 注意：线程只能启动一次，所以每次输入命令4都要创建新的线程实例。

完整代码示例 #

import threading
import time

def loop(stop_event):
    while not stop_event.is_set():
        print("ON")
        print("OFF")
        time.sleep(0.1)  # 加短延迟避免输出刷屏

if __name__ == "__main__":
    current_thread = None
    current_timer = None

    try:
        while True:
            print(">>", end='')
            command = int(input())

            if command == 4:
                print("Insert time:")
                duration = float(input())
                # 先终止之前可能存在的线程和定时器
                if current_thread and current_thread.is_alive():
                    if current_timer:
                        current_timer.cancel()
                    stop_event.set()
                    current_thread.join()
                # 创建新的终止信号、线程和定时器
                stop_event = threading.Event()
                current_thread = threading.Thread(target=loop, args=(stop_event,))
                current_timer = threading.Timer(duration, stop_event.set)
                # 启动线程和定时器
                current_thread.start()
                current_timer.start()
            elif command == 5:
                # 示例：退出主线程
                print("Exiting main thread...")
                # 清理子线程
                if current_thread and current_thread.is_alive():
                    stop_event.set()
                    current_thread.join()
                break
            elif command == 1:
                print("执行命令1的逻辑")
            elif command == 2:
                print("执行命令2的逻辑")
            elif command == 3:
                print("执行命令3的逻辑")
            else:
                print("未知命令，请输入1-5")
    except ValueError:
        print("请输入有效的数字命令")

代码说明 #

• loop函数接收stop_event参数，每次循环都会检查stop_event.is_set()，一旦返回True就退出循环，线程自然终止。
• current_thread和current_timer变量用于跟踪当前运行的子线程和定时器，避免重复启动多个线程。如果用户再次输入4，会先终止之前的线程再启动新的。
• threading.Timer会在指定的duration秒后调用stop_event.set()，触发子线程的退出逻辑。
• 主线程始终处于等待用户输入的循环中，不会被子线程或定时器阻塞，保持响应状态。

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