从 subprocess.communicate() 读取流输入

Posted 2023-02-23

【中文标题】从 subprocess.communicate() 读取流输入

【英文标题】：Read streaming input from subprocess.communicate()

【发布时间】：2011-02-12 12:53:48

【问题描述】：

我正在使用 Python 的 subprocess.communicate() 从运行大约一分钟的进程中读取标准输出。

如何以流方式打印出该进程的stdout 的每一行，以便我可以看到生成的输出，但在继续之前仍会阻止进程终止？

subprocess.communicate() 似乎一次性提供所有输出。

【问题讨论】：

相关：Getting realtime output using subprocess

【参考方案1】：

如果您想要一种非阻塞方法，请不要使用process.communicate()。如果将subprocess.Popen() 参数stdout 设置为PIPE，则可以从process.stdout 读取并检查进程是否仍在使用process.poll() 运行。

【讨论】：

non-blocking approach is not straightforward

【参考方案2】：

请注意，我认为J.F. Sebastian's method (below) 更好。

---

这是一个简单的例子（不检查错误）：

import subprocess
proc = subprocess.Popen('ls',
                       shell=True,
                       stdout=subprocess.PIPE,
                       )
while proc.poll() is None:
    output = proc.stdout.readline()
    print output,

如果ls 结束得太快，那么while 循环可能会在您读取所有数据之前结束。

您可以通过这种方式在标准输出中捕获余数：

output = proc.communicate()[0]
print output,

【讨论】：

这个方案是否会成为python文档所指的缓冲区阻塞问题的牺牲品？

@Heinrich，缓冲区阻塞问题我不太了解。我相信（仅通过谷歌搜索）只有在 while 循环内不从 stdout（和 stderr？）读取时才会出现此问题。所以我觉得上面的代码还可以，但是不能肯定。

这实际上确实遇到了阻塞问题，几年前我一直没有解决 readline 会阻塞的麻烦，直到它得到一个换行符，即使 proc 已经结束。我不记得解决方案，但我认为它与在工作线程上进行读取以及循环 while proc.poll() is None: time.sleep(0) 或类似的东西有关。基本上-您需要确保输出换行符是该过程执行的最后一件事（因为您不能给解释器时间再次循环），或者您需要做一些“花哨的”事情。

@Heinrich：Alex Martelli 在这里写了关于如何避免僵局的文章：***.com/questions/1445627/…

缓冲区阻塞比有时听起来更简单：父块等待子块退出 + 子块等待父块读取并释放通信管道中已满的一些空间 = 死锁。就是这么简单。管道越小，发生的可能性就越大。

【参考方案3】：

我相信以流方式从进程中收集输出的最简单方法是这样的：

import sys
from subprocess import *
proc = Popen('ls', shell=True, stdout=PIPE)
while True:
    data = proc.stdout.readline()   # Alternatively proc.stdout.read(1024)
    if len(data) == 0:
        break
    sys.stdout.write(data)   # sys.stdout.buffer.write(data) on Python 3.x

readline() 或 read() 函数只应在进程终止后在 EOF 上返回一个空字符串 - 否则如果没有可读取的内容，它将阻塞（readline() 包括换行符，因此在空行上，它返回“\n”）。这避免了在循环之后需要尴尬的最终 communicate() 调用。

在行很长的文件上read() 可能更适合减少最大内存使用量 - 传递给它的数字是任意的，但排除它会导致一次读取整个管道输出，这可能是不可取的。

【讨论】：

data = proc.stdout.read() 阻塞，直到 all 数据被读取。您可能会将它与可以更早返回的os.read(fd, maxsize) 混淆（只要有任何数据可用）。

你是对的，我错了。但是，如果将合理数量的字节作为参数传递给read()，那么它工作正常，同样readline() 工作正常，只要最大行长度是合理的。相应地更新了我的答案。

【参考方案4】：

在子进程刷新其标准输出缓冲区后，逐行获取子进程的输出：

#!/usr/bin/env python2
from subprocess import Popen, PIPE

p = Popen(["cmd", "arg1"], stdout=PIPE, bufsize=1)
with p.stdout:
    for line in iter(p.stdout.readline, b''):
        print line,
p.wait() # wait for the subprocess to exit

iter() 用于在写入行后立即读取行，以解决方法the read-ahead bug in Python 2。

如果子进程的标准输出在非交互模式下使用块缓冲而不是行缓冲（这会导致输出延迟，直到子进程的缓冲区已满或被子进程显式刷新），那么您可以尝试强制使用pexpect, pty modules 或unbuffer, stdbuf, script utilities 的无缓冲输出，请参阅Q: Why not just use a pipe (popen())?

---

这是 Python 3 代码：

#!/usr/bin/env python3
from subprocess import Popen, PIPE

with Popen(["cmd", "arg1"], stdout=PIPE, bufsize=1,
           universal_newlines=True) as p:
    for line in p.stdout:
        print(line, end='')

注意：不像 Python 2 那样输出子进程的字节串； Python 3 使用文本模式（cmd 的输出使用locale.getpreferredencoding(False) 编码解码）。

【讨论】：

b'' 是什么意思？

b'' 是 Python 2.7 和 Python 3 中的 bytes 文字。

@JinghaoShi: bufsize=1 如果你也写（使用p.stdin）到子进程，它可以帮助避免死锁，同时执行交互式（pexpect-like）交换——假设子进程本身没有缓冲问题。如果您只是阅读，那么正如我所说，差异仅在于性能：如果不是这样，那么您能否提供一个最小的完整代码示例来显示它？

@ealeon：是的。它需要可以read stdout/stderr concurrently 的技术，除非您将stderr 合并到stdout 中（通过将stderr=subprocess.STDOUT 传递给Popen()）。另请参阅，threading 或 asyncio solutions 链接在那里。

@saulspatz 如果stdout=PIPE 没有捕获输出（您仍然可以在屏幕上看到它），那么您的程序可能会打印到 stderr 或直接打印到终端。要合并 stdout&stderr，请传递 stderr=subprocess.STDOUT（请参阅我之前的评论）。要捕获直接打印到您的 tty 的输出，您可以use pexpect, pty solutions.。这是more complex code example。

【参考方案5】：

如果您只是想实时传递输出，很难比这更简单：

import subprocess

# This will raise a CalledProcessError if the program return a nonzero code.
# You can use call() instead if you don't care about that case.
subprocess.check_call(['ls', '-l'])

请参阅docs for subprocess.check_call()。

如果你需要处理输出，当然，循环它。但是，如果您不这样做，请保持简单。

编辑：J.F. Sebastian 指出 stdout 和 stderr 参数的默认值都传递给 sys.stdout 和 sys.stderr，如果 sys.stdout 和 sys.stderr 会失败。 stderr 已被替换（例如，用于捕获测试中的输出）。

【讨论】：

如果sys.stdout 或sys.stderr 被替换为没有真正fileno() 的类文件对象，它将不起作用。如果sys.stdout、sys.stderr不被替换，那就更简单了：subprocess.check_call(args)。

谢谢！我意识到替换 sys.stdout/stderr 的变幻莫测，但不知何故从未意识到，如果你省略参数，它会将 stdout 和 stderr 传递到正确的位置。我喜欢call() 而不是check_call()，除非我想要CalledProcessError。

python -mthis: “错误永远不应该默默地传递。除非明确地沉默。” 这就是为什么示例代码应该更喜欢check_call() call().

嘿。我结束的很多程序call()ing 在非错误条件下返回非零错误代码，因为它们很糟糕。所以在我们的例子中，非零错误代码实际上并不是错误。

是的。有诸如grep 之类的程序即使没有错误也可能返回非零退出状态——它们是异常。默认情况下，退出状态为零表示成功。

【参考方案6】：

myCommand="ls -l"
cmd=myCommand.split()
# "universal newline support" This will cause to interpret \n, \r\n and \r     equally, each as a newline.
p = subprocess.Popen(cmd, stderr=subprocess.PIPE, universal_newlines=True)
while True:    
    print(p.stderr.readline().rstrip('\r\n'))

【讨论】：

最好解释一下你的解决方案是做什么的，只是为了让人们更好地理解

您应该考虑使用shlex.split(myCommand) 而不是myCommand.split()。它也尊重引用参数中的空格。

【参考方案7】：

添加另一个 python3 解决方案并进行一些小改动：

允许您捕获 shell 进程的退出代码（我在使用 with 构造时无法获取退出代码） 还可以实时输出标准错误

import subprocess
import sys
def subcall_stream(cmd, fail_on_error=True):
    # Run a shell command, streaming output to STDOUT in real time
    # Expects a list style command, e.g. `["docker", "pull", "ubuntu"]`
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, bufsize=1, universal_newlines=True)
    for line in p.stdout:
        sys.stdout.write(line)
    p.wait()
    exit_code = p.returncode
    if exit_code != 0 and fail_on_error:
        raise RuntimeError(f"Shell command failed with exit code exit_code. Command: `cmd`")
    return(exit_code)