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

Question

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

我希望使用Python来读取文件，输出第一个文件的 header ，然后通过管道将尾部进行排序。我尝试过将此作为概念证明:

#!/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() 可能也可以工作；否则，在关闭 stdin 后，您可能需要调用 sorter.wait() 让它完成) .

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

subprocess.Popen的.communicate()方法使用线程或 select每当您传递 _communicate 时，模块原语本身(取决于操作系统支持；实现位于 various subprocess.PIPE methods here 下)适用于多个标准 handle ；这就是你必须这样做的方式。