在固定时间内接收多个输入

本文介绍了在固定时间内接收多个输入的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

我正在使用Python 3，我想编写一个程序，该程序在一定时间内要求多个用户输入.这是我的尝试:

I'm using Python 3 and I wanted to code a program that asks for multiple user inputs for a certain amount of time. Here is my attempt at that:

from threading import Timer ## def timeup(): global your_time your_time = False return your_time ## timeout = 5 your_Time = True t = Timer(timeout, timeup) t.start() ## while your_time == True: input() t.cancel() print('Stop typing!')

问题是，即使时间到了，代码仍然等待输入.我希望循环在时间用完时完全停止.我该怎么做呢?谢谢！

The problem is, the code still waits for an input even if the time is up. I would like the loop to stop exactly when the time runs out. How do I do this? Thank you!

推荐答案

此解决方案是与平台无关的，并且立即会中断键入以通知现有的超时.不必等到用户按下ENTER键就可以发现发生了超时.除了及时通知用户之外，这还可以确保在进一步处理进入的超时后不再输入任何内容.

This solution is platform-independent and immediately interrupts typing to inform about an existing timeout. It doesn't have to wait until the user hits ENTER to find out a timeout occured. Besides informing the user just-in-time this ensures no input after the timeout stepped in is further processed.

功能

• 与平台无关(Unix/Windows).
• 仅StdLib，没有外部依赖项.
• 仅线程，没有子进程.
• 超时立即中断.
• 在超时时清除提示的关闭.
• 在整个时间段内可以无限输入.
• 易于扩展的PromptManager类.
• 程序可能会在超时后恢复，可能会在不重新启动程序的情况下多次运行提示实例.

此答案使用线程管理器实例，该实例在一个 单独的提示线程和MainThread.管理器线程检查超时，并将输入从提示线程转发到父线程.通过这种设计，可以轻松修改MainThread，使其无需阻塞(更改_poll来替换阻塞的queue.get()).

This answer uses a threaded manager instance, which mediates between a separate prompting thread and the MainThread. The manager-thread checks for timeout and forwards inputs from the prompt-thread to the parent-thread. This design enables easy modification in case MainThread would need to be non-blocking (changes in _poll to replace blocking queue.get()).

在超时时，管理器线程要求ENTER继续并使用 threading.Event实例，以确保提示线程在关闭之前关闭 继续.在特定方法的文档文本中查看更多详细信息:

On timeout the manager thread asks for ENTER to continue and uses an threading.Event instance to assure the prompt-thread shuts down before continuing. See further details in the doc-texts of the specific methods:

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() # synchronization for shutdown self._echoed = Event() # synchronization for terminal output 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() # prevent intermixed display 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() # finalize 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

请注意，在尝试输入世界"时，此处会弹出超时消息.

Note the timeout-message here popped up during the attempt of typing "World".