我有以下似乎有效的代码，用于在python中通过子进程将管道链接在一起，同时逐行读取/写入它们（无需预先使用communicate()）。该代码仅调用Unix命令（mycmd），读取其输出，然后将其写入另一个Unix命令（next_cmd）的stdin，然后将最后一个命令的输出重定向到文件。
# some unix command that uses a pipe: command "a"
# writes to stdout and "b" reads it and writes to stdout
mycmd = "a | b"
mycmd_proc = subprocess.Popen(mycmd, shell=True,
stdin=sys.stdin,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
# nextCmd reads from stdin, and I'm passing it mycmd's output
next_cmd = "nextCmd -stdin"
output_file = open(output_filename, "w")
next_proc = subprocess.Popen(next_cmd, shell=True,
stdin=subprocess.PIPE,
stdout=output_file)
for line in iter(mycmd.stdout.readline, ''):
# do something with line
# ...
# write it to next command
next_proc.stdin.write(line)
### If I wanted to call another command here that passes next_proc output
### line by line to another command, would I need
### to call next_proc.communicate() first?
next_proc.communicate()
output_file.close()
这似乎可行，并且仅在命令末尾调用
communicate()。我正在尝试扩展此代码以添加另一个命令，以便您可以执行以下操作：
mycmd1 | mycmd2 | mycmd3 > some_file
含义：逐行，从Python读取mycmd1的输出，处理该行，将其馈送到mycmd2，读取mycmd2的输出，并逐行处理它，并将其馈送到mycmd3，后者又将其输出放入
some_file。这是否可能，或者必然会以死锁/阻塞/未刷新的缓冲区结束？请注意，我不只是将三个unix命令作为管道调用，因为我想在它们之间介入Python，并在将其输入到下一个命令之前逐行对每个命令的输出进行后处理。我想避免调用communication并将所有输出加载到内存中，而是想逐行解析它。谢谢。
最佳答案
这应该处理任意数量的命令：
import sys
import subprocess
def processFirst(out):
return out
def processSecond(out):
return out
def processThird(out):
return out
commands = [("a|b", processFirst), ("nextCmd -stdin", processSecond), ("thirdCmd", processThird)]
previous_output = None
for cmd,process_func in commands:
if previous_output is None:
stdin = sys.stdin
else:
stdin = subprocess.PIPE
proc = subprocess.Popen(cmd, shell=True,
stdin = stdin,
stdout = subprocess.PIPE)
if previous_output is not None:
proc.stdin.write(previous_output)
out,err = proc.communicate()
out = process_func(out)
previous_output = out
只需将要运行的任何命令以及应处理其输出的函数添加到命令列表中即可。最后一个命令的输出最终将在循环末尾的
previous_output中。为了避免任何死锁/缓冲/等问题，您只需使用
proc.communicate()运行每个命令即可完成操作，这将返回输出（而不是像示例中那样直接读取）。然后，您可以将其输入到下一条命令中，然后再执行它，以此类推。编辑：刚注意到您不想在前面使用
communicate()，并且想要逐行做出反应。我将编辑我的答案以解决该问题This answer提供了有关如何使用
select.select()从管道逐行读取而不阻塞的示例。以下是针对您的特定情况使用它的示例：
import sys
import subprocess
import select
import os
class LineReader(object):
def __init__(self, fd, process_func):
self._fd = fd
self._buf = ''
self._process_func = process_func
self.next_proc = None
def fileno(self):
return self._fd
def readlines(self):
data = os.read(self._fd, 4096)
if not data:
# EOF
if self.next_proc is not None:
self.next_proc.stdin.close()
return None
self._buf += data
if '\n' not in data:
return []
tmp = self._buf.split('\n')
tmp_lines, self._buf = tmp[:-1], tmp[-1]
lines = []
for line in tmp_lines:
lines.append(self._process_func(line))
if self.next_proc is not None:
self.next_proc.stdin.write("%s\n" % lines[-1])
return lines
def processFirst(line):
return line
def processSecond(line):
return line
def processThird(line):
return line
commands = [("a|b", processFirst), ("nextCmd -stdin", processSecond), ("thirdCmd", processThird)]
readers = []
previous_reader = None
for cmd,process_func in commands:
if previous_reader is None:
stdin = sys.stdin
else:
stdin = subprocess.PIPE
proc = subprocess.Popen(cmd, shell=True,
stdin = stdin,
stdout = subprocess.PIPE)
if previous_reader is not None:
previous_reader.next_proc = proc
previous_reader = LineReader(proc.stdout.fileno(), process_func)
readers.append(previous_reader)
while readers:
ready,_,_ = select.select(readers, [], [], 10.0)
for stream in ready:
lines = stream.readlines()
if lines is None:
readers.remove(stream)
关于python - 通过子过程在Python中逐行写入/读取管道来逐行链接，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/15189611/