You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何终止包含input()的线程?无需等待输入即可关闭程序

问题解决:让程序可随时响应关闭操作而无需等待输入完成

问题根源

你的代码中,输入线程被input()调用阻塞,即便主线程检测到关闭快捷键,也必须等输入完成才能结束该线程,导致程序无法及时响应关闭指令。sys.exit()无效是因为它仅终止当前线程,输入线程仍处于阻塞状态。

解决方案

我们可以通过非阻塞输入+线程事件通知的方式解决,既保留输入功能,又能让程序随时响应关闭操作。以下是修改后的代码:

import threading
import keyboard as kb
import sys
import select

# 用于通知输入线程退出的事件
exit_event = threading.Event()
input_output = None

def turn_off():
    return kb.is_pressed('shift') and kb.is_pressed('1')

def Specjal_input():
    global input_output
    print("请输入一个数字:", end='', flush=True)
    while not exit_event.is_set():
        # 每隔0.1秒检查一次是否有输入,同时响应退出信号
        ready, _, _ = select.select([sys.stdin], [], [], 0.1)
        if ready:
            # 读取输入并去除换行符
            input_output = sys.stdin.readline().strip()
            break
    # 若因退出信号终止,清空输入缓冲区避免残留内容
    if exit_event.is_set():
        try:
            sys.stdin.read()
        except:
            pass

running = True
Input_Thread = threading.Thread(target=Specjal_input)
Input_Thread.start()

while running:
    if turn_off():
        exit_event.set()
        running = False
    elif input_output is not None:
        running = False

if input_output is not None:
    print(f'\n输入的数字:{input_output}')
else:
    print("\n程序正在关闭,输入已取消")
Input_Thread.join()

代码说明

  • 使用select.select()实现非阻塞输入,每隔0.1秒检查一次输入状态,期间可以响应退出事件
  • exit_event线程事件用于安全通知输入线程终止,避免强制终止线程带来的资源泄漏问题
  • 检测到关闭快捷键时,设置exit_event,输入线程会在下一次循环检查时退出
  • 清空输入缓冲区,防止残留输入影响后续程序运行

备选方案(简单但不推荐)

如果追求极简实现,可以通过强制向输入线程发送异常来打断input()调用,但这种方式可能导致资源未正确释放:

import threading
import keyboard as kb
import ctypes

def turn_off():
    return kb.is_pressed('shift') and kb.is_pressed('1')

input_output = None

def Specjal_input():
    global input_output
    try:
        input_output = input('请输入一个数字:')
    except KeyboardInterrupt:
        pass

running = True
Input_Thread = threading.Thread(target=Specjal_input)
Input_Thread.start()

while running:
    if turn_off():
        if Input_Thread.is_alive():
            # 向输入线程发送KeyboardInterrupt异常
            ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(Input_Thread.ident), ctypes.py_object(KeyboardInterrupt))
        running = False
    elif input_output is not None:
        running = False

if input_output is not None:
    print(f'\n输入的数字:{input_output}')
else:
    print("\n程序正在关闭,输入已取消")
Input_Thread.join()

内容的提问来源于stack exchange,提问作者Loleczkowo

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.19 13:10:05