我使用subprocess模块启动子流程并连接到其输出流(标准输出)。我希望能够对其标准输出执行非阻塞读取。是否有一种方法使.readline非阻塞或在我调用.readline之前检查流上是否有数据?我希望这是可移植的，或者至少可以在Windows和Linux下工作。
下面是我现在怎么做的(如果没有数据可用，它会阻塞在.readline上):
p = subprocess.Popen('myprogram.exe', stdout = subprocess.PIPE)
output_str = p.stdout.readline()
当前回答
使用select & read(1)。
import subprocess #no new requirements
def readAllSoFar(proc, retVal=''):
while (select.select([proc.stdout],[],[],0)[0]!=[]):
retVal+=proc.stdout.read(1)
return retVal
p = subprocess.Popen(['/bin/ls'], stdout=subprocess.PIPE)
while not p.poll():
print (readAllSoFar(p))
readline()例如:
lines = ['']
while not p.poll():
lines = readAllSoFar(p, lines[-1]).split('\n')
for a in range(len(lines)-1):
print a
lines = readAllSoFar(p, lines[-1]).split('\n')
for a in range(len(lines)-1):
print a
其他回答
我的问题有点不同，因为我想从正在运行的进程中收集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新手。
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之上，那么这真的不会有多大帮助。希望这能对其他读者有所帮助，即使它不适用于您的特定应用程序。
下面是我的代码，用于捕获子进程ASAP的每个输出，包括部分行。它同时泵浦，stdout和stderr的顺序几乎是正确的。
在Python 2.7 linux和windows上测试并正确工作。
#!/usr/bin/python
#
# Runner with stdout/stderr catcher
#
from sys import argv
from subprocess import Popen, PIPE
import os, io
from threading import Thread
import Queue
def __main__():
if (len(argv) > 1) and (argv[-1] == "-sub-"):
import time, sys
print "Application runned!"
time.sleep(2)
print "Slept 2 second"
time.sleep(1)
print "Slept 1 additional second",
time.sleep(2)
sys.stderr.write("Stderr output after 5 seconds")
print "Eol on stdin"
sys.stderr.write("Eol on stderr\n")
time.sleep(1)
print "Wow, we have end of work!",
else:
os.environ["PYTHONUNBUFFERED"]="1"
try:
p = Popen( argv + ["-sub-"],
bufsize=0, # line-buffered
stdin=PIPE, stdout=PIPE, stderr=PIPE )
except WindowsError, W:
if W.winerror==193:
p = Popen( argv + ["-sub-"],
shell=True, # Try to run via shell
bufsize=0, # line-buffered
stdin=PIPE, stdout=PIPE, stderr=PIPE )
else:
raise
inp = Queue.Queue()
sout = io.open(p.stdout.fileno(), 'rb', closefd=False)
serr = io.open(p.stderr.fileno(), 'rb', closefd=False)
def Pump(stream, category):
queue = Queue.Queue()
def rdr():
while True:
buf = stream.read1(8192)
if len(buf)>0:
queue.put( buf )
else:
queue.put( None )
return
def clct():
active = True
while active:
r = queue.get()
try:
while True:
r1 = queue.get(timeout=0.005)
if r1 is None:
active = False
break
else:
r += r1
except Queue.Empty:
pass
inp.put( (category, r) )
for tgt in [rdr, clct]:
th = Thread(target=tgt)
th.setDaemon(True)
th.start()
Pump(sout, 'stdout')
Pump(serr, 'stderr')
while p.poll() is None:
# App still working
try:
chan,line = inp.get(timeout = 1.0)
if chan=='stdout':
print "STDOUT>>", line, "<?<"
elif chan=='stderr':
print " ERROR==", line, "=?="
except Queue.Empty:
pass
print "Finish"
if __name__ == '__main__':
__main__()
选择模块帮助您确定下一个有用的输入在哪里。
但是，您几乎总是更喜欢使用单独的线程。一个是阻塞读取stdin，另一个是阻塞你不想阻塞的地方。
下面是一个在python中支持非阻塞读和后台写的模块:
https://pypi.python.org/pypi/python-nonblock
提供一个函数，
nonblock_read将从流中读取数据，如果可用，否则返回一个空字符串(或None，如果流在另一端关闭，并且所有可能的数据都已读取)
你也可以考虑python-subprocess2模块，
https://pypi.python.org/pypi/python-subprocess2
这将添加到子流程模块。所以在subprocess返回的对象上。Popen”被添加了一个额外的方法，runInBackground。这将启动一个线程并返回一个对象，该对象将在写入stdout/stderr时自动填充，而不会阻塞主线程。
享受吧!
