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

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

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

特征

• 平台无关（Unix / Windows）。
• 仅StdLib，无外部依赖项。
• 仅线程，无子进程。
• 超时立即中断。
• 超时时清除提示器的关闭。
• 在时间跨度内可以无限输入。
• 易于扩展的promptManager类。
• 程序可能会在超时后恢复，可能会在不重新启动程序的情况下多次运行提示实例。

此答案使用线程管理器实例，该实例在单独的提示线程和MainThread之间进行中介。管理器线程检查超时，并将输入从提示线程转发到父线程。通过这种设计，可以轻松修改MainThread，使其无需阻塞（进行更改

_poll

以替换阻塞

queue.get()

）。

在超时时，管理器线程要求ENTER继续，并使用一个

threading.Event

实例来确保提示线程在继续之前关闭。在特定方法的文档文本中查看更多详细信息：

from threading import Thread, Eventfrom queue import Queue, Emptyimport timeSENTINEL = 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()

示例输出：

>$ Helloyou typed: Hello>$ WorTime is out! Press ENTER to continue.Process finished with exit pre 0

请注意，在尝试键入“世界”时，此处会d出超时消息。