所以，我有一个小控制台应用程序我正在运行一个Web刮的过程，我希望能够给它的控制台命令中旬执行，以控制它。 要做到这一点，我需要某种形式的非阻塞键盘输入的，因为程序可能会由于意外的错误自我终止，而我不希望一些不绝如缕约而等待输入发生终止时。
我有以下的已散列出来：
import threading
import time
import queue
input_queue = queue.Queue()
command_input_event = threading.Event()
def kbdListener():
global input_queue, command_input_event
kbdInput = ''
while kbdInput.lower() not in ['quit', 'exit', 'stop']:
kbdInput = input("> ")
input_queue.put(kbdInput)
command_input_event.set()
input_queue.join()
listener = threading.Thread(target=kbdListener)
listener.start()
stop = False
while not stop:
if command_input_event.is_set():
while not input_queue.empty():
command = input_queue.get()
if command.lower() in ['quit', 'exit', 'stop']:
print('Stopping')
while not input_queue.empty():
input_queue.get()
input_queue.task_done()
input_queue.task_done()
stop = True
break
else:
print('Command "{}" received and processed'.format(command))
input_queue.task_done()
我的问题是，就行了while not stop:会有我的程序，即确定主循环已经终止正在检查另一个条件。 如果这种可能性是发生那么主线程将停止，但后台listener线程仍然会等待输入; 这种情况我试图避免的。
我不依赖于这种方法，所以如果有是获取无阻塞输入一些替代方法，那么我会开到他的意见为好。
所以以后这方面的一些工作，我想出了以下内容，这允许后台线程输入，然后由前台线程处理，但完全无阻塞和更新实时的用户类型。
import threading
import queue
import sys
from msvcrt import getch, kbhit
"""
Optional extra that I found necessary to get ANSI commands working on windows:
import colorama
colorama.init()
"""
class ClassWithNonBlockingInput:
def __init__(self):
self.command_queue = queue.Queue()
self.command_available_event = threading.Event()
self.stop_event = threading.Event()
def run(self):
main_loop = threading.Thread(target=self.main_loop)
main_loop.start()
listener = threading.Thread(target=self.keyboard_listener)
listener.start()
stop = False
while main_loop.is_alive() and not stop:
if self.command_available_event.is_set():
while not self.command_queue.empty():
command = self.command_queue.get()
#Process command here, long jobs should be on a seperate thread.
self.command_queue.task_done()
self.command_available_event.clear()
def main_loop(self):
#Main processing loop, may set self.stop_event at some point to terminate early.
pass
def keyboard_listener(self):
line = []
line_changed = False
while not self.stop_event.is_set():
while kbhit():
c = getch()
if c == b'\x00' or c == b'\xe0':
#This is a special function key such as F1 or the up arrow.
#There is a second identifier character to clear/identify the key.
id = getch()
#Process the control key by sending commands as necessary.
elif c == b'\x08':
#Backspace character, remove last character.
if len(line) > 0:
line = line[:-1]
line_changed = True
elif c == b'\x7f':
#ctrl-backspace, remove characters until last space.
while len(line) > 0 and line[-1] != b' ':
line = line[:-1]
line_changed = True
elif c == b'\x1b':
#Escacpe key, process as necessary.
pass
elif c == b'\r':
#Enter key, send command to main thread.
print()
command = b''.join(line).decode('utf-8')
self.command_queue.put(command)
self.command_available_event.set()
self.command_queue.join()
line = []
line_changed = True
else:
#Append all other characters to the current line.
#There may be other special keys which need to be considered,
# this is left as an exercise for the reader :P
line.append(c)
line_changed = True
if line_changed:
#Clear the current output line
print('\033[2K', end='\r')
print(b''.join(line).decode('utf-8'), end='')
sys.stdout.flush()
line_changed = False
这应该给任何人谁遇到在未来的这个问题，在这个问题一个很好的开端绊倒。
