“3 路”Python 子流程管道：将标准输出和标准错误发送到两个不同的进程？

Question

“标准”子流程管道技术（例如http://docs.python.org/2/library/subprocess.html#replacing-shell-pipeline）能否“升级”为两个管道？

# How about
p1 = Popen(["cmd1"], stdout=PIPE, stderr=PIPE)
p2 = Popen(["cmd2"], stdin=p1.stdout)
p3 = Popen(["cmd3"], stdin=p1.stderr)
p1.stdout.close()  # Allow p1 to receive a SIGPIPE if p2 exits.
p1.stderr.close()
#p2.communicate()  # or p3.communicate()?

好的，这实际上是一个不同的用例，但最近的起点似乎是管道示例。顺便说一句，“正常”管道中的 p2.communicate() 如何驱动 p1？这是供参考的正常管道：

# From Python docs
output=`dmesg | grep hda`
# becomes
p1 = Popen(["dmesg"], stdout=PIPE)
p2 = Popen(["grep", "hda"], stdin=p1.stdout, stdout=PIPE)
p1.stdout.close()  # Allow p1 to receive a SIGPIPE if p2 exits.
output = p2.communicate()[0]

我想我最终对communicate() 支持什么样的“过程图”（或者可能只是树？）感兴趣，但我们将把一般情况留到另一天。

更新：这是基线功能。没有communicate()，创建2个线程从p1.stdout和p2.stdout读取。在主进程中，通过 p1.stdin.write() 注入输入。问题是我们是否可以只使用communicate() 来驱动一个1-source，2-sink 图

Answer 1

你可以使用 bash 的 process substitution:

from subprocess import check_call

check_call("cmd1 > >(cmd2) 2> >(cmd3)", shell=True, executable="/bin/bash")

它将cmd1 的标准输出重定向到cmd2，并将cmd1 的标准错误重定向到cmd3。

如果您不想使用bash，那么您问题中的代码应该可以正常工作，例如：

#!/usr/bin/env python
import sys
from subprocess import Popen, PIPE
from textwrap import dedent

# generate some output on stdout/stderr
source = Popen([sys.executable, "-c", dedent("""
    from __future__ import print_function
    import sys
    from itertools import cycle
    from string import ascii_lowercase

    for i, c in enumerate(cycle(ascii_lowercase)):
        print(c)
        print(i, file=sys.stderr)
""")], stdout=PIPE, stderr=PIPE)

# convert input to upper case
sink = Popen([sys.executable, "-c", dedent("""
    import sys

    for line in sys.stdin:
        sys.stdout.write(line.upper())
""")], stdin=source.stdout)
source.stdout.close() # allow source to receive SIGPIPE if sink exits

# square input
sink_stderr = Popen([sys.executable, "-c", dedent("""
   import sys

   for line in sys.stdin:
       print(int(line)**2)
""")], stdin=source.stderr)
source.stderr.close() # allow source to receive SIGPIPE if sink_stderr exits

sink.communicate()
sink_stderr.communicate()
source.wait()

Answer 2

这里的解决方案是创建几个后台线程，它们从一个进程读取输出，然后将其写入多个进程的输入：

targets = [...] # list of processes as returned by Popen()
while True:
    line = p1.readline()
    if line is None: break
    for p in targets:
        p.stdin.write(line)