0

我正在尝试运行一个程序，该程序在工作完成时接受输入。我查看了几种表格，并查看了文档。我在 Debian 中运行它，我知道我可以使用这个getch 函数来接收字符而无需按回车键。为了打破它，这就是我试图在我的无限while循环中实现的

  • 接受输入（线程在这里对我不起作用
  • 将输入放入队列
  • 如果没有正在运行的作业，则以队列前面的项目作为变量启动作业

我也在运行线程模块来执行另一条指令。有什么办法可以做到这一点吗？

更新：这是我迄今为止尝试过的：

首先，我尝试使用线程模块中的计时器来阻止它等待，就像这样。

def getchnow():    
        def time_up():
            answer= None
            print 'time up...'

    wait = Timer(5,time_up) # x is amount of time in seconds
    wait.start()
    try:
            print "enter answer below"
            answer = getch()
    except Exception:
            print 'pass\n'
            answer = None

    if answer != True:   # it means if variable have somthing 
            wait.cancel()       # time_up will not execute(so, no skip)
    return answer
line = getchnow()
#Add line variable to queue
#Do stuff with queue

这里的问题是它仍在等待用户输入。

然后我尝试将 getch 函数放入另一个线程。

q = Queue.Queue
q.put(getch())  
if q.get() != True:   # it means if variable have somthing
    line = q.get()
    #Add line variable to queue
#Do stuff with queue

这种尝试不会让我做任何事情。

4

2 回答 2

0

我阅读了这个链接的更多内容，并且在底部有一个我想要的实现。

我将 select 模块用于 Linux 上的非阻塞实现。如果没有收到输入，这将在（此处为 5 秒）超时。在线程中使用时特别有用，因此 getch 调用是非阻塞的，并允许线程干净地退出

# This class gets a single character input from the keyboard
class _GetchUnix:
    def __init__(self):
        import tty, sys
        from select import select
    def __call__(self):
        import sys, tty, termios
        from select import select
        fd = sys.stdin.fileno()
        old_settings = termios.tcgetattr(fd)
        try:
                tty.setraw(sys.stdin.fileno())
                [i, o, e] = select([sys.stdin.fileno()], [], [], 2)
                if i: 
                ch=sys.stdin.read(1)
                else: 
                ch='' 
        finally:
                    termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
            return ch
getch = _GetchUnix()
# End Class
于 2013-03-09T00:49:37.083 回答
0

我也用过[i, o, e] = select([sys.stdin.fileno()], [], [], 2)，但听说它可能无法在 Windows 上运行。如果有人仍然需要多线程、非阻塞输入示例：

import threading
import sys
import time

bufferLock=threading.Lock()
inputBuffer=[]

class InputThread(threading.Thread):
    def run(self):
        global inputBuffer
        print("starting input")
        while True:
            line=sys.stdin.readline()
            bufferLock.acquire()
            inputBuffer.insert(0,line)
            bufferLock.release()

input_thread=InputThread()
input_thread.start()
while True:
    time.sleep(4)
    bufferLock.acquire()
    if len(inputBuffer)>0:
        print("Popping: "+inputBuffer.pop())
    bufferLock.release()
于 2015-07-02T15:37:00.100 回答