Python读取子进程的标准输出和标准错误并保持顺序

Question

Python读取子进程的标准输出和标准错误并保持顺序

64

我有一个 Python 子进程，我试图从中读取输出和错误流。目前我已经使其工作，但我只能在完成从 stdout 读取后才能从 stderr 读取。代码如下：

process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout_iterator = iter(process.stdout.readline, b"")
stderr_iterator = iter(process.stderr.readline, b"")

for line in stdout_iterator:
    # Do stuff with line
    print line

for line in stderr_iterator:
    # Do stuff with line
    print line

如您所见，stderr 循环只能在 stdout 循环完成后开始。我该如何修改代码才能按正确的顺序从两个循环中读取行？

澄清一下： 我仍然需要能够确定一行是来自 stdout 还是 stderr，因为它们会在我的代码中被不同地处理。

- Leah Sapan

3

链接：在类似终端的实时环境中运行命令并分别获取其标准输出和标准错误输出要求能够在类似终端的实时环境中运行命令，并实时获取其标准输出和标准错误输出。 - jfs

7个回答

22

这里有一个基于选择器的解决方案，但它保留了顺序并流式传输可变长度的字符（即使是单个字符）。

诀窍在于使用read1()，而不是read()。

import selectors
import subprocess
import sys

p = subprocess.Popen(
    ["python", "random_out.py"], stdout=subprocess.PIPE, stderr=subprocess.PIPE
)

sel = selectors.DefaultSelector()
sel.register(p.stdout, selectors.EVENT_READ)
sel.register(p.stderr, selectors.EVENT_READ)

while True:
    for key, _ in sel.select():
        data = key.fileobj.read1().decode()
        if not data:
            exit()
        if key.fileobj is p.stdout:
            print(data, end="")
        else:
            print(data, end="", file=sys.stderr)

如果您需要一个测试程序，请使用此程序。

import sys
from time import sleep


for i in range(10):
    print(f" x{i} ", file=sys.stderr, end="")
    sleep(0.1)
    print(f" y{i} ", end="")
    sleep(0.1)

- Dev Aggarwal

1

注意：1-它在Windows上无法工作。2-它不会保留顺序（只是使您更不可能注意到顺序错误）。请参见我的答案下的相关评论。 - jfs

@DevAggarwal 我没有尝试它，因为我找到了另一种解决我的问题的替代方案，不需要合并流。 - shouldsee

#python3.8：使用read1()有时会截断我的长输出。通过增加缓冲区大小（即：read1(size=1000000)）可能可以解决此问题，但这可能会禁用“读取并返回数据直到达到EOF”的功能，如https://docs.python.org/3/library/io.html?highlight=read1#io.BufferedIOBase.read1中所述。最终我改用了read()。 - Nikolas

这看起来非常酷，但似乎不能保持顺序 :( - Tim

不要忘记加上 flush=True，否则在这个例子中你将无法获得实时输出。 - SodaCris

显示剩余4条评论

10

这适用于Python3（3.6）：

    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, 
                         stderr=subprocess.PIPE, universal_newlines=True)
    # Read both stdout and stderr simultaneously
    sel = selectors.DefaultSelector()
    sel.register(p.stdout, selectors.EVENT_READ)
    sel.register(p.stderr, selectors.EVENT_READ)
    ok = True
    while ok:
        for key, val1 in sel.select():
            line = key.fileobj.readline()
            if not line:
                ok = False
                break
            if key.fileobj is p.stdout:
                print(f"STDOUT: {line}", end="")
            else:
                print(f"STDERR: {line}", end="", file=sys.stderr)

- Deepak Yadav

9

进程将数据写入不同的管道的顺序在写操作后丢失。

无法确定stdout是否在stderr之前被写入。

您可以尝试以非阻塞的方式同时从多个文件描述符读取数据，只要数据可用即可，但这只能最小化顺序不正确的概率。

此程序应该演示此问题：

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import select
import subprocess

testapps={
    'slow': '''
import os
import time
os.write(1, 'aaa')
time.sleep(0.01)
os.write(2, 'bbb')
time.sleep(0.01)
os.write(1, 'ccc')
''',
    'fast': '''
import os
os.write(1, 'aaa')
os.write(2, 'bbb')
os.write(1, 'ccc')
''',
    'fast2': '''
import os
os.write(1, 'aaa')
os.write(2, 'bbbbbbbbbbbbbbb')
os.write(1, 'ccc')
'''
}

def readfds(fds, maxread):
    while True:
        fdsin, _, _ = select.select(fds,[],[])
        for fd in fdsin:
            s = os.read(fd, maxread)
            if len(s) == 0:
                fds.remove(fd)
                continue
            yield fd, s
        if fds == []:
            break

def readfromapp(app, rounds=10, maxread=1024):
    f=open('testapp.py', 'w')
    f.write(testapps[app])
    f.close()

    results={}
    for i in range(0, rounds):
        p = subprocess.Popen(['python', 'testapp.py'], stdout=subprocess.PIPE
                                                     , stderr=subprocess.PIPE)
        data=''
        for (fd, s) in readfds([p.stdout.fileno(), p.stderr.fileno()], maxread):
            data = data + s
        results[data] = results[data] + 1 if data in results else 1

    print 'running %i rounds %s with maxread=%i' % (rounds, app, maxread)
    results = sorted(results.items(), key=lambda (k,v): k, reverse=False)
    for data, count in results:
        print '%03i x %s' % (count, data)


print
print "=> if output is produced slowly this should work as whished"
print "   and should return: aaabbbccc"
readfromapp('slow',  rounds=100, maxread=1024)

print
print "=> now mostly aaacccbbb is returnd, not as it should be"
readfromapp('fast',  rounds=100, maxread=1024)

print
print "=> you could try to read data one by one, and return"
print "   e.g. a whole line only when LF is read"
print "   (b's should be finished before c's)"
readfromapp('fast',  rounds=100, maxread=1)

print
print "=> but even this won't work ..."
readfromapp('fast2', rounds=100, maxread=1)

并输出类似以下的内容：

=> if output is produced slowly this should work as whished
   and should return: aaabbbccc
running 100 rounds slow with maxread=1024
100 x aaabbbccc

=> now mostly aaacccbbb is returnd, not as it should be
running 100 rounds fast with maxread=1024
006 x aaabbbccc
094 x aaacccbbb

=> you could try to read data one by one, and return
   e.g. a whole line only when LF is read
   (b's should be finished before c's)
running 100 rounds fast with maxread=1
003 x aaabbbccc
003 x aababcbcc
094 x abababccc

=> but even this won't work ...
running 100 rounds fast2 with maxread=1
003 x aaabbbbbbbbbbbbbbbccc
001 x aaacbcbcbbbbbbbbbbbbb
008 x aababcbcbcbbbbbbbbbbb
088 x abababcbcbcbbbbbbbbbb

- Jörg Schulz

1

在这里使用if not s:而不是if len(s) == 0:。在这里使用while fds:而不是while True: ... if fds == []: break。使用results = collections.defaultdict(int); ...; results[data]+=1而不是results = {}; ...; results[data] = results[data] + 1 if data in results else 1。 - jfs

或者使用results = collections.Counter(); ...; results[data]+=1; ...; for data, count in results.most_common():。 - jfs

你可以使用 data = b''.join([s for _, s in readfds(...)])。 - jfs

你应该关闭管道，避免依赖垃圾回收来释放父进程中的文件描述符，并调用 p.wait() 显式地回收子进程。 - jfs

请注意：如果有多个并行进程，则“慢”可能不足以获得所需的输出。 - jfs

5

来自https://docs.python.org/3/library/subprocess.html#using-the-subprocess-module

如果您希望捕获并合并两个流到一个流中，请使用stdout=PIPE和stderr=STDOUT而不是capture_output。

因此，最简单的解决方案是：

process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
stdout_iterator = iter(process.stdout.readline, b"")

for line in stdout_iterator:
    # Do stuff with line
    print line

- Itay Bianco

1

这个程序不会分别读取流，而是将stderr合并到stdout中。 - Stan Svec

2

我知道这个问题很老，但是这个答案可能会帮助到那些在寻找类似解决方案时偶然发现此页面的人，所以我还是要发布它。
我编写了一个简单的Python片段，将任意数量的管道合并为一个管道。当然，如上所述，无法保证顺序，但我认为这是Python中最接近的方法。
它为每个管道生成一个线程，逐行读取它们并将它们放入队列（FIFO）。主线程通过循环遍历队列，产生每一行。

import threading, queue
def merge_pipes(**named_pipes):
    r'''
    Merges multiple pipes from subprocess.Popen (maybe other sources as well).
    The keyword argument keys will be used in the output to identify the source
    of the line.

    Example:
    p = subprocess.Popen(['some', 'call'],
                         stdin=subprocess.PIPE,
                         stdout=subprocess.PIPE,
                         stderr=subprocess.PIPE)
    outputs = {'out': log.info, 'err': log.warn}
    for name, line in merge_pipes(out=p.stdout, err=p.stderr):
        outputs[name](line)

    This will output stdout to the info logger, and stderr to the warning logger
    '''

    # Constants. Could also be placed outside of the method. I just put them here
    # so the method is fully self-contained
    PIPE_OPENED=1
    PIPE_OUTPUT=2
    PIPE_CLOSED=3

    # Create a queue where the pipes will be read into
    output = queue.Queue()

    # This method is the run body for the threads that are instatiated below
    # This could be easily rewritten to be outside of the merge_pipes method,
    # but to make it fully self-contained I put it here
    def pipe_reader(name, pipe):
        r"""
        reads a single pipe into the queue
        """
        output.put( ( PIPE_OPENED, name, ) )
        try:
            for line in iter(pipe.readline,''):
                output.put( ( PIPE_OUTPUT, name, line.rstrip(), ) )
        finally:
            output.put( ( PIPE_CLOSED, name, ) )

    # Start a reader for each pipe
    for name, pipe in named_pipes.items():
        t=threading.Thread(target=pipe_reader, args=(name, pipe, ))
        t.daemon = True
        t.start()

    # Use a counter to determine how many pipes are left open.
    # If all are closed, we can return
    pipe_count = 0

    # Read the queue in order, blocking if there's no data
    for data in iter(output.get,''):
        code=data[0]
        if code == PIPE_OPENED:
            pipe_count += 1
        elif code == PIPE_CLOSED:
            pipe_count -= 1
        elif code == PIPE_OUTPUT:
            yield data[1:]
        if pipe_count == 0:
            return

- Marten Jacobs

0

这对我有用（在Windows上）： https://github.com/waszil/subpiper

from subpiper import subpiper

def my_stdout_callback(line: str):
    print(f'STDOUT: {line}')

def my_stderr_callback(line: str):
    print(f'STDERR: {line}')

my_additional_path_list = [r'c:\important_location']

retcode = subpiper(cmd='echo magic',
                   stdout_callback=my_stdout_callback,
                   stderr_callback=my_stderr_callback,
                   add_path_list=my_additional_path_list)

- waszil

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- jfs · Accepted Answer

如果子进程在stderr上产生足够多的输出（在我的Linux机器上大约为100KB），则您问题中的代码可能会出现死锁。

有一个communicate()方法，可以分别从stdout和stderr读取：

from subprocess import Popen, PIPE

process = Popen(command, stdout=PIPE, stderr=PIPE)
output, err = process.communicate()

如果您需要在子进程仍在运行时读取流，则可采用线程的便携解决方案（未经测试）：

from subprocess import Popen, PIPE
from threading import Thread
from Queue import Queue # Python 2

def reader(pipe, queue):
    try:
        with pipe:
            for line in iter(pipe.readline, b''):
                queue.put((pipe, line))
    finally:
        queue.put(None)

process = Popen(command, stdout=PIPE, stderr=PIPE, bufsize=1)
q = Queue()
Thread(target=reader, args=[process.stdout, q]).start()
Thread(target=reader, args=[process.stderr, q]).start()
for _ in range(2):
    for source, line in iter(q.get, None):
        print "%s: %s" % (source, line),

请参见: