<p>下面的代码用于跨平台提示<code>input()</code><strong>一次
超时。超时时,它会要求用户按Enter键,以便提示器-
线程可以完全关闭。你知道吗</p>
<p>删除影响<code>_prompter_exit</code>的部分将允许在超时时恢复,而无需用户按Enter键,代价是在整个进程退出之前保持提示线程的活动状态。你知道吗</p>
<p>仅仅像@Mad physicast建议的那样加入一个超时提示线程是行不通的,因为它不会解除<code>input()</code>调用本身的阻塞。如果不使用特定于操作系统的库,那么如果不最终提供某种形式的输入,就无法唤醒等待<code>input()</code>的线程。你知道吗</p>
<p>代码是我答案的简化版本,允许在给定的时间跨度内进行多个输入:</p>
<p><a href="https://stackoverflow.com/a/53180738/9059420">Taking in multiple inputs for a fixed time in Python</a></p>
<pre><code>from threading import Thread, enumerate, Event
from queue import Queue, Empty
import time
SENTINEL = None
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._prompter_exit = Event()
def run(self):
"""Run worker-thread. Start prompt-thread, fetch passed
input from in_queue and forward it to `._poll()` in MainThread.
If timeout occurs before user-input, enqueue SENTINEL to
unblock `.get()` in `._poll()`.
"""
self.prompter.start()
try:
txt = self._in_queue.get(timeout=self.timeout)
except Empty:
self._out_queue.put(SENTINEL)
print(f"\n[{time.ctime()}] Please press Enter to continue.")
# without usage of _prompter_exit() and Enter, the
# prompt-thread would stay alive until the whole program ends
self._prompter_exit.wait()
else:
self._out_queue.put(txt)
def start(self):
"""Start manager-thread."""
super().start()
return self._poll()
def _prompter(self):
"""Prompting target function for execution in prompter-thread."""
self._in_queue.put(input(f"[{time.ctime()}] >$ "))
self._prompter_exit.set()
def _poll(self):
"""Get forwarded inputs from the manager-thread executing `run()`
and process them in the parent-thread.
"""
msg = self._out_queue.get()
self.join()
return msg
</code></pre>
<p>用于演示:</p>
<pre><code>if __name__ == '__main__':
pm = PromptManager(timeout=5)
msg = pm.start()
print(f"User input: {msg}")
for i in range(3):
print(f"[{time.ctime()}] Do something else. "
f"Alive threads:{[t.name for t in enumerate()]}")
time.sleep(1)
</code></pre>
<p>运行触发超时:</p>
<pre class="lang-none prettyprint-override"><code>[Tue Nov 26 20:50:47 2019] >$
[Tue Nov 26 20:50:52 2019] Please press Enter to continue.
User input: None
[Tue Nov 26 20:50:57 2019] Do something else. Alive threads:['MainThread']
[Tue Nov 26 20:50:58 2019] Do something else. Alive threads:['MainThread']
[Tue Nov 26 20:50:59 2019] Do something else. Alive threads:['MainThread']
Process finished with exit code 0
</code></pre>
<p>使用用户输入及时运行:</p>
<pre class="lang-none prettyprint-override"><code>[Tue Nov 26 20:51:16 2019] >$ Hello
User input: Hello
[Tue Nov 26 20:51:19 2019] Do something else. Alive threads:['MainThread']
[Tue Nov 26 20:51:20 2019] Do something else. Alive threads:['MainThread']
[Tue Nov 26 20:51:21 2019] Do something else. Alive threads:['MainThread']
Process finished with exit code 0
</code></pre>