如何从`stdin`创建非阻塞连续读取？

Question

我有一个进程,它是这样创建的:

p = subprocess.Popen(args   = './myapp',
                     stdin  = subprocess.PIPE,
                     stdout = subprocess.PIPE,
                     universal_newlines=True)

后来,我试着写p的stdin:

p.stdin.write('my message\n')

该myapp过程具有以下设置:

q = queue.Queue()
def get_input():
    for line in iter(sys.stdin.readline, ''):
        q.put(line)
    sys.stdin.close()

threading.Thread(name   = 'input-getter',
                 target = get_input).start()

它正在尝试不断读取新行,如下所示:

try:
    print('input:', q.get_nowait())
except Empty:
    print('no input')

不幸的是,子进程从不接收任何消息.当然,当我使用时:

p.communicate('my message\n')

子进程临危消息,但正如所料,communicate方法关闭p的stdin,所以没有更多的沟通正在进行.

Answer 1

p = subprocess.Popen(args   = './myapp',
                     stdin  = subprocess.PIPE,
                     stdout = subprocess.PIPE,
                     universal_newlines=True)

while p.poll() is None:
    data = p.stdout.readline()

这将创建一个非阻塞的进程读取,直到进程退出.但是,这里有一些注意事项需要注意.例如,如果你也管道stderr,但不读取它..那么你很可能填充一两个缓冲区,你仍然会挂起程序.因此,在手动执行操作时,请务必确保清除任何缓冲区I/O.

一个更好的选择是尽可能使用select.epoll(),这只能在unix系统上使用,但会给你带来更好的性能和错误处理:)

epoll = select.epoll()
epoll.register(p.stdout.fileno(), select.EPOLLHUP) # Use select.EPOLLIN for stdin.

for fileno, event in epoll.poll(1):
    if fileno == p.stdout.fileno():
        # ... Do something ...

注:请记住,只要一个过程预计输入时,它通常是通过指示此stdout,所以你仍然会注册STDOUT使用select.epoll,以检查"等待输入".您可以注册select.EPOLLIN以检查输入是否已给出,但我几乎看不到重点,因为请记住,您选择输入到您应该已经知道的过程中的是"正在发生".

检查进程是否需要输入

您可以使用select.epoll以上示例检查进程是否正在等待输入而不阻止应用程序执行.但还有更好的选择.

Pexpect是一个能够很好地完成它并与之合作的库SSH.

它与子进程有点不同,但可能是一个不错的选择.

让subprocess.popen使用SSH

我将重定向到另一个问题+答案,如果这是你所追求的(因为SSH将以stdin受保护的方式产生一个).

Python + SSH密码验证(没有外部库或公钥/私钥)？