python - 如何将输出多路复用到 Python 中的操作系统文件描述符？

Question

subprocess.Popen 机制使用底层文件描述符而不是类文件对象来写入其 stdout/stderr。我需要同时捕获 stdout 和 stderr 并将它们显示到控制台。

我如何创建 Popen 可以使用的文件描述符来允许我执行此操作？

Answer 1

只是一些上下文:subprocess uses the raw file descriptors您指定的stdin、stdout、stderr 对象，因为it passes them down to POSIX .如果您使用 subprocess.PIPE，那么它将使用 os.pipe() 创建一个新管道。此外，Popen.communicate 一直读取到流的末尾，如果您想将数据传输到其他地方，这可能是不可取的。

因为您想将输出打印到stdout，所以我假设它是文本输出。您需要在 Popen 中使用 encoding、errors 或 universal_newlines 来为 subprocess将文件视为文本(参见 docs )。

import subprocess

p = subprocess.Popen(
    '/usr/bin/whoami',
    stdout=subprocess.PIPE,  # Control stdout
    universal_newlines=True  # Files opened in text mode
)

# Pipe the data somewhere else too, e.g.: a log file
with open('subprocess.log', 'w') as logfile:
    # p.poll() returns the return code when `p` exits
    while p.poll() is None:
        line = p.stdout.readline()
        # one to our stdout (readline includes the \n)
        print(line, end='')
        # one to the logfile
        logfile.write(line)

可以使用相同的技术来操纵 stderr，例如，通过将 file=sys.stderr 传递给 print。请注意，您也可以通过直接传递您自己的 stdin 进行管道传输:

subprocess.Popen('/usr/bin/whoami', stdin=sys.stdin, stdout=subprocess.PIPE, ...)

毕竟，标准流只是包装文件描述符。如果读取到行尾不适合您期望的输出类型，您可以只读取一个非常短的缓冲区。

同时使用stderr 和stdout

如果您同时需要 stdout 和 stderr，就会遇到一次只能读取一个的问题。
一种可能性是使用 os.set_blocking 使管道成为非阻塞的，以便任何 read 方法在没有数据时立即返回。这允许您在流之间切换。
另一种可能性是让两个单独的线程处理 stdout 和 stderr；但是有一种更简单的方法可以通过 aysncio module 来实现。 :

import asyncio
import sys

PROCESS_PATH = '/bin/mixed_output'

class MultiplexProtocol(asyncio.SubprocessProtocol):
    def __init__(self, exit_future):
        self.exit_future = exit_future

    def pipe_data_received(self, fd, data):
        if fd == sys.stdout.fileno():
            print(data.decode('utf-8'), file=sys.stdout, end='')
        elif fd == sys.stderr.fileno():
            print(data.decode('utf-8'), file=sys.stderr, end='')

    def process_exited(self):
        self.exit_future.set_result(True)


async def launch_subprocess(loop):
    # Future marking the end of the process
    exit_future = asyncio.Future(loop=loop)
    # Use asyncio's subprocess
    create_subp = loop.subprocess_exec(
        lambda: MultiplexProtocol(exit_future),
        PROCESS_PATH,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        stdin=None
    )
    transport, protocol = await create_subp
    await exit_future
    # Close the pipes
    transport.close()


loop = asyncio.get_event_loop()
loop.run_until_complete(launch_subprocess(loop))

这比在主机进程中不断循环以将数据传输到其他流要少得多，因为 MultiplexProtocol.pipe_data_received 仅在需要时调用。