python - "3-way"Python 子进程管道 : send stdout and stderr to two different processes?

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() 注入(inject)输入。问题是我们是否可以仅使用 communicate() 驱动一个 1-source, 2-sink graph

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()