python - Python 中子进程 PIPE 的非阻塞读取，一次一个字节

Question

我在这个问题中实现了代码的变体:

A non-blocking read on a subprocess.PIPE in Python

尝试从这个虚拟程序中实时读取输出test.py:

import time, sys

print "Hello there"
for i in range(100):
    time.sleep(0.1)
    sys.stdout.write("\r%d"%i)
    sys.stdout.flush()
print
print "Go now or I shall taunt you once again!"

另一个问题的变化是调用程序必须逐字符读取，而不是逐行读取，因为虚拟程序 test.py 使用 \r。所以这里是:

import sys,time
from subprocess import PIPE, Popen
from threading  import Thread

try:
    from Queue import Queue, Empty
except ImportError:
    from queue import Queue, Empty  # Python 3.x

ON_POSIX = 'posix' in sys.builtin_module_names

def enqueue_output(out, queue):
    while True:
        buffersize = 1
        data = out.read(buffersize)
        if not data:
            break
        queue.put(data)
    out.close()

p = Popen(sys.executable + " test.py", stdout=PIPE, bufsize=1, close_fds=ON_POSIX)
q = Queue()
t = Thread(target=enqueue_output, args=(p.stdout, q))
t.daemon = True # Thread dies with the program
t.start()

while True:
    p.poll()
    if p.returncode:
        break
    # Read line without blocking
    try:
        char = q.get_nowait()
        time.sleep(0.1)
    except Empty:
        pass
    else: # Got line
        sys.stdout.write(char)
        sys.stdout.flush()

print "left loop"
sys.exit(0)

有两个问题

• 它永远不会退出 - p.returncode 永远不会返回值，并且循环不会离开。我该如何修复它？
• 真的很慢!有没有办法在不增加缓冲区大小的情况下提高效率？

Answer 1

正如 @Markku K. 指出的，您应该使用 bufsize=0 一次读取一个字节。

您的代码不需要非阻塞读取。您可以简化它:

import sys
from functools import partial
from subprocess import Popen, PIPE

p = Popen([sys.executable, "test.py"], stdout=PIPE, bufsize=0)
for b in iter(partial(p.stdout.read, 1), b""):
    print b # it should print as soon as `sys.stdout.flush()` is called
            # in the test.py
p.stdout.close()
p.wait()

注意:一次读取 1 个字节的效率非常低。

此外，一般来说，可能有 a block-buffering issue有时可以使用 pexpect, pty modules 来解决或unbuffer, stdbuf, script command-line utilities .

对于 Python 进程，您可以使用 -u 标志来强制 stdin、stdout、stderr 流的非缓冲(二进制层)。