我使用subprocess模块启动子流程并连接到其输出流(标准输出)。我希望能够对其标准输出执行非阻塞读取。是否有一种方法使.readline非阻塞或在我调用.readline之前检查流上是否有数据?我希望这是可移植的，或者至少可以在Windows和Linux下工作。
下面是我现在怎么做的(如果没有数据可用，它会阻塞在.readline上):
p = subprocess.Popen('myprogram.exe', stdout = subprocess.PIPE)
output_str = p.stdout.readline()
当前回答
我的问题有点不同，因为我想从正在运行的进程中收集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新手。
其他回答
试试asyncproc模块。例如:
import os
from asyncproc import Process
myProc = Process("myprogram.app")
while True:
# check to see if process has ended
poll = myProc.wait(os.WNOHANG)
if poll != None:
break
# print any new output
out = myProc.read()
if out != "":
print out
该模块负责S.Lott建议的所有线程。
我有最初提问者的问题，但不希望调用线程。我将Jesse的解决方案与来自管道的直接read()和我自己的行读取缓冲处理程序混合在一起(然而，我的子进程- ping -总是写入<系统页面大小的整行)。我通过读入一个gobject-registered io watch来避免忙碌的等待。这些天我通常在gobject MainLoop中运行代码以避免线程。
def set_up_ping(ip, w):
# run the sub-process
# watch the resultant pipe
p = subprocess.Popen(['/bin/ping', ip], stdout=subprocess.PIPE)
# make stdout a non-blocking file
fl = fcntl.fcntl(p.stdout, fcntl.F_GETFL)
fcntl.fcntl(p.stdout, fcntl.F_SETFL, fl | os.O_NONBLOCK)
stdout_gid = gobject.io_add_watch(p.stdout, gobject.IO_IN, w)
return stdout_gid # for shutting down
观察者是
def watch(f, *other):
print 'reading',f.read()
return True
主程序建立一个ping，然后调用gobject邮件循环。
def main():
set_up_ping('192.168.1.8', watch)
# discard gid as unused here
gobject.MainLoop().run()
任何其他工作都附加到gobject中的回调。
我的问题有点不同，因为我想从正在运行的进程中收集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新手。
为什么要打扰线程和队列? 与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__()
使用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
