根据J.F. Sebastian的答案和其他几个来源，我组合了一个简单的子流程管理器。它提供了请求的非阻塞读取，以及并行运行几个进程。它不使用任何特定于操作系统的调用(我知道)，因此应该在任何地方工作。

它可以从pypi获得，所以只需pip安装shelljob。参考项目页面的示例和完整的文档。

You can do this really easily in Twisted. Depending upon your existing code base, this might not be that easy to use, but if you are building a twisted application, then things like this become almost trivial. You create a ProcessProtocol class, and override the outReceived() method. Twisted (depending upon the reactor used) is usually just a big select() loop with callbacks installed to handle data from different file descriptors (often network sockets). So the outReceived() method is simply installing a callback for handling data coming from STDOUT. A simple example demonstrating this behavior is as follows:

from twisted.internet import protocol, reactor

class MyProcessProtocol(protocol.ProcessProtocol):

    def outReceived(self, data):
        print data

proc = MyProcessProtocol()
reactor.spawnProcess(proc, './myprogram', ['./myprogram', 'arg1', 'arg2', 'arg3'])
reactor.run()

Twisted文档在这方面有一些很好的信息。

如果您围绕Twisted构建整个应用程序，它可以与其他进程(本地或远程)进行异步通信，就像这样非常优雅。另一方面，如果您的程序不是构建在Twisted之上，那么这真的不会有多大帮助。希望这能对其他读者有所帮助，即使它不适用于您的特定应用程序。

我的问题有点不同，因为我想从正在运行的进程中收集stdout和stderr，但最终是一样的，因为我想在小部件生成时在小部件中呈现输出。

我不希望使用队列或额外的线程来解决许多建议的解决方案，因为执行运行另一个脚本并收集其输出这样的常见任务应该不需要它们。

在阅读了建议的解决方案和python文档后，我解决了以下实现的问题。是的，它只适用于POSIX，因为我正在使用select函数调用。

我同意，对于这样一个常见的脚本任务，文档是令人困惑的，实现是尴尬的。我相信旧版本的python对Popen有不同的默认值和不同的解释，所以造成了很多困惑。这似乎对Python 2.7.12和3.5.2都很有效。

关键是将bufsize=1设置为行缓冲，然后universal_newlines=True处理为文本文件而不是二进制文件，这似乎成为设置bufsize=1时的默认值。

class workerThread(QThread):
   def __init__(self, cmd):
      QThread.__init__(self)
      self.cmd = cmd
      self.result = None           ## return code
      self.error = None            ## flag indicates an error
      self.errorstr = ""           ## info message about the error

   def __del__(self):
      self.wait()
      DEBUG("Thread removed")

   def run(self):
      cmd_list = self.cmd.split(" ")   
      try:
         cmd = subprocess.Popen(cmd_list, bufsize=1, stdin=None
                                        , universal_newlines=True
                                        , stderr=subprocess.PIPE
                                        , stdout=subprocess.PIPE)
      except OSError:
         self.error = 1
         self.errorstr = "Failed to execute " + self.cmd
         ERROR(self.errorstr)
      finally:
         VERBOSE("task started...")
      import select
      while True:
         try:
            r,w,x = select.select([cmd.stdout, cmd.stderr],[],[])
            if cmd.stderr in r:
               line = cmd.stderr.readline()
               if line != "":
                  line = line.strip()
                  self.emit(SIGNAL("update_error(QString)"), line)
            if cmd.stdout in r:
               line = cmd.stdout.readline()
               if line == "":
                  break
               line = line.strip()
               self.emit(SIGNAL("update_output(QString)"), line)
         except IOError:
            pass
      cmd.wait()
      self.result = cmd.returncode
      if self.result < 0:
         self.error = 1
         self.errorstr = "Task terminated by signal " + str(self.result)
         ERROR(self.errorstr)
         return
      if self.result:
         self.error = 1
         self.errorstr = "exit code " + str(self.result)
         ERROR(self.errorstr)
         return
      return

ERROR, DEBUG和VERBOSE仅仅是将输出打印到终端的宏。

这个解决方案在我看来是99.99%有效的，因为它仍然使用了阻塞readline函数，所以我们假设子进程很好，输出完整的行。

我欢迎反馈，以改进解决方案，因为我仍然是Python新手。

在现代Python中，情况要好得多。

下面是一个简单的子程序"hello.py":

#!/usr/bin/env python3

while True:
    i = input()
    if i == "quit":
        break
    print(f"hello {i}")

以及一个与之交互的程序:

import asyncio


async def main():
    proc = await asyncio.subprocess.create_subprocess_exec(
        "./hello.py", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE
    )
    proc.stdin.write(b"bob\n")
    print(await proc.stdout.read(1024))
    proc.stdin.write(b"alice\n")
    print(await proc.stdout.read(1024))
    proc.stdin.write(b"quit\n")
    await proc.wait()


asyncio.run(main())

打印出来:

b'hello bob\n'
b'hello alice\n'

请注意，实际的模式(几乎所有前面的回答，包括这里和相关的问题)是将子进程的stdout文件描述符设置为非阻塞，然后在某种选择循环中轮询它。当然，现在这个循环是由asyncio提供的。

根据J.F. Sebastian的答案和其他几个来源，我组合了一个简单的子流程管理器。它提供了请求的非阻塞读取，以及并行运行几个进程。它不使用任何特定于操作系统的调用(我知道)，因此应该在任何地方工作。

它可以从pypi获得，所以只需pip安装shelljob。参考项目页面的示例和完整的文档。

为什么要打扰线程和队列? 与readline()不同，BufferedReader.read1()不会阻塞等待\r\n，如果有任何输出进入，它会尽快返回。

#!/usr/bin/python
from subprocess import Popen, PIPE, STDOUT
import io

def __main__():
    try:
        p = Popen( ["ping", "-n", "3", "127.0.0.1"], stdin=PIPE, stdout=PIPE, stderr=STDOUT )
    except: print("Popen failed"); quit()
    sout = io.open(p.stdout.fileno(), 'rb', closefd=False)
    while True:
        buf = sout.read1(1024)
        if len(buf) == 0: break
        print buf,

if __name__ == '__main__':
    __main__()