这里没有一个答案能满足我所有的需求。

没有用于标准输出的线程(也没有队列等) 非阻塞，因为我需要检查其他事情正在进行 使用PIPE，因为我需要做很多事情，例如流输出，写入日志文件，并返回输出的字符串副本。

一点背景知识:我使用ThreadPoolExecutor来管理一个线程池，每个线程启动一个子进程并并发地运行它们。(在Python2.7中，但这应该在更新的3中工作。X也是)。我不希望只使用线程来收集输出，因为我希望有尽可能多的可用线程用于其他事情(一个20个进程的池将使用40个线程来运行;1用于进程线程，1用于stdout…如果你想要stderr我猜)

我在这里剥离了很多异常，所以这是基于在生产中工作的代码。希望我没有在复制粘贴过程中破坏它。同时，非常欢迎反馈!

import time
import fcntl
import subprocess
import time

proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

# Make stdout non-blocking when using read/readline
proc_stdout = proc.stdout
fl = fcntl.fcntl(proc_stdout, fcntl.F_GETFL)
fcntl.fcntl(proc_stdout, fcntl.F_SETFL, fl | os.O_NONBLOCK)

def handle_stdout(proc_stream, my_buffer, echo_streams=True, log_file=None):
    """A little inline function to handle the stdout business. """
    # fcntl makes readline non-blocking so it raises an IOError when empty
    try:
        for s in iter(proc_stream.readline, ''):   # replace '' with b'' for Python 3
            my_buffer.append(s)

            if echo_streams:
                sys.stdout.write(s)

            if log_file:
                log_file.write(s)
    except IOError:
        pass

# The main loop while subprocess is running
stdout_parts = []
while proc.poll() is None:
    handle_stdout(proc_stdout, stdout_parts)

    # ...Check for other things here...
    # For example, check a multiprocessor.Value('b') to proc.kill()

    time.sleep(0.01)

# Not sure if this is needed, but run it again just to be sure we got it all?
handle_stdout(proc_stdout, stdout_parts)

stdout_str = "".join(stdout_parts)  # Just to demo

我确信这里有额外的开销，但在我的情况下这不是一个问题。从功能上讲，它满足了我的需要。我唯一没有解决的问题是，为什么这对于日志消息非常有效，但我看到一些打印消息在稍后同时显示。

如果有人想同时使用线程从stdout和stderr读取，这是我想到的:

import threading
import subprocess
import Queue

class AsyncLineReader(threading.Thread):
    def __init__(self, fd, outputQueue):
        threading.Thread.__init__(self)

        assert isinstance(outputQueue, Queue.Queue)
        assert callable(fd.readline)

        self.fd = fd
        self.outputQueue = outputQueue

    def run(self):
        map(self.outputQueue.put, iter(self.fd.readline, ''))

    def eof(self):
        return not self.is_alive() and self.outputQueue.empty()

    @classmethod
    def getForFd(cls, fd, start=True):
        queue = Queue.Queue()
        reader = cls(fd, queue)

        if start:
            reader.start()

        return reader, queue


process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
(stdoutReader, stdoutQueue) = AsyncLineReader.getForFd(process.stdout)
(stderrReader, stderrQueue) = AsyncLineReader.getForFd(process.stderr)

# Keep checking queues until there is no more output.
while not stdoutReader.eof() or not stderrReader.eof():
   # Process all available lines from the stdout Queue.
   while not stdoutQueue.empty():
       line = stdoutQueue.get()
       print 'Received stdout: ' + repr(line)

       # Do stuff with stdout line.

   # Process all available lines from the stderr Queue.
   while not stderrQueue.empty():
       line = stderrQueue.get()
       print 'Received stderr: ' + repr(line)

       # Do stuff with stderr line.

   # Sleep for a short time to avoid excessive CPU use while waiting for data.
   sleep(0.05)

print "Waiting for async readers to finish..."
stdoutReader.join()
stderrReader.join()

# Close subprocess' file descriptors.
process.stdout.close()
process.stderr.close()

print "Waiting for process to exit..."
returnCode = process.wait()

if returnCode != 0:
   raise subprocess.CalledProcessError(returnCode, command)

我只是想分享这个，因为我在这个问题上试图做类似的事情，但没有一个答案能解决我的问题。希望它能帮助到某些人!

请注意，在我的用例中，外部进程杀死了Popen()所使用的进程。

这至少在Python3.4中是有效的

import subprocess

process = subprocess.Popen(cmd_list, stdout=subprocess.PIPE)
for line in process.stdout:
    print(line.decode().strip())

这个PoC不断地从进程中读取输出，并可以在需要时访问。只有最后一个结果被保留，所有其他输出都被丢弃，因此防止了PIPE的内存不足:

import subprocess
import time
import threading
import Queue


class FlushPipe(object):
    def __init__(self):
        self.command = ['python', './print_date.py']
        self.process = None
        self.process_output = Queue.LifoQueue(0)
        self.capture_output = threading.Thread(target=self.output_reader)

    def output_reader(self):
        for line in iter(self.process.stdout.readline, b''):
            self.process_output.put_nowait(line)

    def start_process(self):
        self.process = subprocess.Popen(self.command,
                                        stdout=subprocess.PIPE)
        self.capture_output.start()

    def get_output_for_processing(self):
        line = self.process_output.get()
        print ">>>" + line


if __name__ == "__main__":
    flush_pipe = FlushPipe()
    flush_pipe.start_process()

    now = time.time()
    while time.time() - now < 10:
        flush_pipe.get_output_for_processing()
        time.sleep(2.5)

    flush_pipe.capture_output.join(timeout=0.001)
    flush_pipe.process.kill()

print_date.py

#!/usr/bin/env python
import time

if __name__ == "__main__":
    while True:
        print str(time.time())
        time.sleep(0.01)

输出:你可以清楚地看到只有大约2.5s的输出间隔。

>>>1520535158.51
>>>1520535161.01
>>>1520535163.51
>>>1520535166.01

简单比复杂好。

操作系统库内置模块系统。您应该执行代码并看到输出。

import os
os.system("python --version")
# Output
"""
Python 3.8.6
0
"""

在版本之后，它也打印返回值为0。

对于试图回答这个问题并从Python脚本中获取标准输出的人来说，请注意Python会缓冲它的标准输出，因此可能需要一段时间才能看到标准输出。

这可以通过在目标脚本中的每个标准输出写入后添加以下内容来纠正:

sys.stdout.flush()