这个解决方案是跨平台的,立即中断打字操作以通知已存在的超时。它不必等到用户按下ENTER键才发现超时发生了。除了及时通知用户外,这还确保了超时后没有输入会被进一步处理。
特性
- 跨平台(Unix / Windows)。
- 仅使用StdLib,无需外部依赖。
- 仅使用线程,无需使用子进程。
- 在超时时立即中断。
- 在超时时清洁关闭提示器。
- 在时间范围内可进行无限输入。
- 易于扩展的PromptManager类。
- 程序可以在超时后恢复,可以多次运行提示器实例而无需重新启动程序。
这个答案使用了一个线程管理器实例,它在一个单独的提示线程和主线程之间进行调解。管理线程检查超时并把来自提示线程的输入转发给父线程。这种设计使得在需要使MainThread非阻塞的情况下容易进行修改(在_poll中更改以取代阻塞的queue.get())。
在超时时,管理线程请求按下ENTER键继续,并使用 threading.Event
实例来确保在继续之前提示线程关闭。有关特定方法的详细信息,请参见文档文本:
from threading import Thread, Event
from queue import Queue, Empty
import time
SENTINEL = object()
class PromptManager(Thread):
def __init__(self, timeout):
super().__init__()
self.timeout = timeout
self._in_queue = Queue()
self._out_queue = Queue()
self.prompter = Thread(target=self._prompter, daemon=True)
self.start_time = None
self._prompter_exit = Event()
self._echoed = Event()
def run(self):
"""Run in worker-thread. Start prompt-thread, fetch passed
inputs from in_queue and check for timeout. Forward inputs for
`_poll` in parent. If timeout occurs, enqueue SENTINEL to
break the for-loop in `_poll()`.
"""
self.start_time = time.time()
self.prompter.start()
while self.time_left > 0:
try:
txt = self._in_queue.get(timeout=self.time_left)
except Empty:
self._out_queue.put(SENTINEL)
else:
self._out_queue.put(txt)
print("\nTime is out! Press ENTER to continue.")
self._prompter_exit.wait()
@property
def time_left(self):
return self.timeout - (time.time() - self.start_time)
def start(self):
"""Start manager-thread."""
super().start()
self._poll()
def _prompter(self):
"""Prompting target function for execution in prompter-thread."""
while self.time_left > 0:
self._in_queue.put(input('>$ '))
self._echoed.wait()
self._echoed.clear()
self._prompter_exit.set()
def _poll(self):
"""Get forwarded inputs from the manager-thread executing `run()`
and process them in the parent-thread.
"""
for msg in iter(self._out_queue.get, SENTINEL):
print(f'you typed: {msg}')
self._echoed.set()
self._echoed.set()
self._prompter_exit.wait()
self.join()
if __name__ == '__main__':
pm = PromptManager(timeout=5)
pm.start()
示例输出:
>$ Hello
you typed: Hello
>$ Wor
Time is out! Press ENTER to continue.
Process finished with exit code 0
请注意,这里的超时提示消息是在尝试输入“World”期间弹出的。
import time
import threading timer = 5 timeout = time.time() + 1 * timer def do_input(): while True: message = input("Message: ") def do_timer(): while time.time() < timeout: continue print("\n计时器已结束!") exit() for i in range(1): thread = threading.Thread(target=do_input) thread.daemon = True thread.start() thread1 = threading.Thread(target=do_timer).start() - Feitan Portortime.sleep
以防止占用 CPU),而exit()
在终端中可以工作,但在 IDE 控制台中无法工作(改为sys.exit()
)。 - Darkonaut