如何在 Python 2.6 中实现复杂的流程管道？答案

【问题标题】：How to implement a complex process pipe in Python 2.6?如何在 Python 2.6 中实现复杂的流程管道？
【发布时间】：2016-04-25 05:53:02
【问题描述】：

我喜欢这个 shell 管道的 Python（2.6，抱歉！）等价物：

$ longrunningprocess | sometextfilter | gzip -c

也就是说，我要调用一个二进制longrunningprocess，通过sometextfilter过滤它的输出，需要得到gzip的输出。

我知道如何使用subprocess 管道，但我需要逐块输出管道（可能使用yield），而不是一次全部使用。例如。这 https://security.openstack.org/guidelines/dg_avoid-shell-true.html 仅适用于一次获取所有输出。

注意，longrunningprocess 和 sometextfilter 都是外部程序，不能用 Python 函数替换。

提前感谢任何提示或示例！

【问题讨论】：

在这种情况下 Python 在做什么？是文本过滤器吗？是压缩包吗？是不是只是在最后收集最终的输出？
不需要通过管道连接到系统 gzip...python 有一个内置的 gzip 接口（通过 gzip 模块）。
而不是复杂的管道，听起来您只需要一个从标准输入读取并将压缩数据写入标准输出（或文件）的python程序。
当然，gzipping可以用Python来完成，但是longrunningprocess和sometextfilter都是C程序，不能用Python重新实现。也就是说，Python 需要读取第一个进程的标准输出，将其通过管道传输到第二个进程的标准输入，最后 gzip 该进程的标准输出。我有点迷失如何在 Python 中做到这一点，而无需等待进程终止，这是无法完成的。
我看到两个独立的问题：How do I use subprocess.Popen to connect multiple processes by pipes? 以及如何以块的形式读取子进程的输出（如果您设置了stdout=PIPE 那么process.stdout 是一个普通的（不可搜索的）文件对象，它具有相应的方法，例如chunk = process.stdout.read(chunk_size))。你试过什么？您的代码有哪些具体问题？

标签： python filter pipe subprocess yield

【解决方案1】：

再次，我认为这很困难，而 Python （应该）很容易。似乎只是连接子流程就可以了：

def get_lines():
    lrp = subprocess.Popen(["longrunningprocess"],
                           stdout=subprocess.PIPE,
                           close_fds=True)
    stf = subprocess.Popen(["sometextfilter"],
                           stdin=lrp.stdout,
                           stdout=subprocess.PIPE,
                           bufsize=1,
                           close_fds=True)

    for l in iter(stf.stdout.readline, ''):
        yield l

    lrp.stdout.close()
    stf.stdout.close()
    stf.stdin.close()

    stf.wait()
    lrp.wait()

[应用了 J.F. Sebastian 的更改。谢谢！]

然后我可以使用 Pythons gzip 进行压缩。

【讨论】：

(1) 你需要lrp.stdout.close() 否则longrunningprocess 可能会在sometextfilter 过早死亡时挂起 (2) close_fds=True 只能在此处的 POSIX 系统上使用 (3) for l in stf.stdout 可能会延迟由于预读错误（使用iter(p.stdout.readline, '') 而不是（4）bufsize=1 在第一次调用中无用（5）关闭管道以避免泄漏文件描述符（6）调用.wait() 以避免僵尸
1- lrp.stdout.close() 可以在Popen(["sometextfilter"], stdin=lrp.stdout,..) 之后立即调用。 2- 我不明白你为什么不在这里使用shell=True（longrunningprocess 的名字表明启动 shell 的开销不是这里的问题）。如果您不向 shell 传递不受信任的输入，则使用 shell 是安全的，即在 Python 源代码中使用带有字符串文字（常量）的 shell=True 是完全可以接受的。 3-如果你坚持使用两个Popen()调用然后切换进程的顺序被调用就像this answer that I've linked earlier

【解决方案2】：

shell 语法针对单行进行了优化，使用它：

#!/usr/bin/env python2
import sys
from subprocess import Popen, PIPE

LINE_BUFFERED = 1
ON_POSIX = 'posix' in sys.builtin_module_names

p = Popen('longrunningprocess | sometextfilter', shell=True,
          stdout=PIPE, bufsize=LINE_BUFFERED, close_fds=ON_POSIX)
with p.stdout:
    for line in iter(p.stdout.readline, ''):
        print line,  # do something with the line
p.wait()

如果您想手动模拟管道：

#!/usr/bin/env python2
import sys
from subprocess import Popen, PIPE

LINE_BUFFERED = 1
ON_POSIX = 'posix' in sys.builtin_module_names

sometextfilter = Popen('sometextfilter', stdin=PIPE, stdout=PIPE,
                       bufsize=LINE_BUFFERED, close_fds=ON_POSIX)
longrunningprocess = Popen('longrunningprocess', stdout=sometextfilter.stdin, 
                           close_fds=ON_POSIX)
with sometextfilter.stdin, sometextfilter.stdout as pipe:
    for line in iter(pipe.readline, ''):
        print line, # do something with the line
sometextfilter.wait()
longrunningprocess.wait()

【讨论】：