使用 subprocess.Popen 将数据流式传输到命令中

Question

我经常需要对包含标题的文件集合进行排序。因为排序取决于标题的内容，所以这个用例比类似的问题更复杂（例如，Is there a way to ignore header lines in a UNIX sort?）。

我希望使用 Python 来读取文件，输出第一个文件的标题，然后将尾部通过管道传输到排序中。我已经尝试将此作为概念证明：

#!/usr/bin/env python

import io
import subprocess
import sys

header_printed = False

sorter = subprocess.Popen(['sort'], stdin=subprocess.PIPE)

for f in sys.argv[1:]:
    fd = io.open(f,'r')
    line = fd.readline()
    if not header_printed:
        print(line)
        header_printed = True
    sorter.communicate(line)

当以header-sort fileA fileB 调用时，fileA 和 fileB 包含类似的行

c   float   int
Y   0.557946     413
F   0.501935     852
F   0.768102     709

我明白了：

# sort file 1
Traceback (most recent call last):
  File "./archive/bin/pipetest", line 17, in <module>
    sorter.communicate(line)
  File "/usr/lib/python2.7/subprocess.py", line 785, in communicate
    self.stdin.write(input)
ValueError: I/O operation on closed file

问题是通信需要一个字符串，并且管道在写入后关闭。这意味着必须将内容完全读入内存。通信不需要生成器（我试过）。

一个更简单的演示是：

>>> import subprocess
>>> p = subprocess.Popen(['tr', 'a-z', 'A-Z'], stdin=subprocess.PIPE)
>>> p.communicate('hello')
HELLO(None, None)
>>> p.communicate('world')
Traceback (most recent call last):
  File "<ipython-input-14-d6873fd0f66a>", line 1, in <module>
    p.communicate('world')
  File "/usr/lib/python2.7/subprocess.py", line 785, in communicate
    self.stdin.write(input)
ValueError: I/O operation on closed file

那么，问题是，在 Python 中将数据流式传输到管道中的正确方法是什么（使用 Popen 或其他方式）？

Answer 1

对于您的特定情况，如果您只为单个标准句柄传递了subprocess.PIPE（在您的情况下为stdin），那么在您的示例中，您可以安全地一遍又一遍地调用sorter.stdin.write(line)。当你写完输出后，调用sorter.stdin.close()，这样sort就知道输入已经完成，它可以执行实际的排序和输出工作（sorter.communicate()不带参数可能也可以工作；否则，在关闭@987654332之后@ 你可能想打电话给sorter.wait() 让它完成）。

如果您需要处理多个管道标准手柄，正确的方法是 threading 为每个管道使用专用线程，必须处理第一个（概念上相对简单，但重量级）并引入了所有令人头疼的线程），或使用select 模块（或在Python 3.4+ 中，selectors 模块），这很难做到正确，但可以（在某些情况下）更有效。最后是creating temporary files for output，所以你可以在进程写入文件时直接写入进程的stdin（因此不会阻塞）；然后，您可以在闲暇时读取文件（请注意，子进程在退出之前不一定会刷新它自己的输出缓冲区，因此在进一步的输入和输出填充并刷新之前，输出可能不会及时到达以响应您的输入缓冲区）。

subprocess.Popen 的.communicate() 方法使用线程或select 模块原语本身（取决于操作系统支持；实现在various _communicate methods here 下），只要您将subprocess.PIPE 传递给多个标准把手；这就是你必须这样做的方式。

Answer 2

直接写入管道即可：

#!/usr/bin/env python2
import fileinput
import subprocess

process = subprocess.Popen(['sort'], stdin=subprocess.PIPE)
with process.stdin as pipe, fileinput.FileInput() as file:
    for line in file:
        if file.isfirstline(): # print header
            print line,
        else: # pipe tails
            pipe.write(line)
process.wait()

Answer 3

您可以使用来自stdin 和stdout 的写入/读取，但是根据您的子进程，您需要一个“刷新机制”让子进程处理您的输入。下面的代码适用于第一部分，但由于它关闭了stdin，它也终止了子进程。如果您使用flush() 更改它，或者如果您可以添加一些尾随字符来推送您的子进程，那么您可以使用它。否则，我建议您看看Multithreading in Python，尤其是pipes。

p=subprocess.Popen(['tr','a-z','A-Z'],stdin=subprocess.PIPE,stdout=subprocess.PIPE)
p.stdin.write("hello\n")
p.stdin.close()
p.stdout.readline()
'HELLO\n'