使用 asyncio 子进程连续输出

Question

我正在使用 asyncio subprocess 来执行子命令。我想查看长时间运行的进程并同时将内容保存到缓冲区以供以后使用。此外，我发现了这个相关的问题（Getting live output from asyncio subprocess），但它主要围绕 ssh 的用例。

asyncio subprocess 文档有一个逐行读取输出的示例，这符合我想要实现的目标。（https://docs.python.org/3/library/asyncio-subprocess.html#examples）

import asyncio
import sys

async def get_date():
    code = 'import datetime; print(datetime.datetime.now())'

    # Create the subprocess; redirect the standard output
    # into a pipe.
    proc = await asyncio.create_subprocess_exec(
        sys.executable, '-c', code,
        stdout=asyncio.subprocess.PIPE)

    # Read one line of output.
    data = await proc.stdout.readline()
    line = data.decode('ascii').rstrip()

    # Wait for the subprocess exit.
    await proc.wait()
    return line

date = asyncio.run(get_date())
print(f"Current date: {date}")

我将此示例改编为以下内容：

async def subprocess_async(cmd, **kwargs):
    cmd_list = shlex.split(cmd)
    proc = await asyncio.create_subprocess_exec(
            *cmd_list,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.STDOUT, **kwargs)

    full_log = ""
    while True:
        buf = await proc.stdout.readline()
        if not buf:
            break
        full_log += buf.decode()
        print(f' {buf.decode().rstrip()}')
    await proc.wait()
    res = subprocess.CompletedProcess(cmd, proc.returncode,  stdout=full_log.encode(), stderr=b'')
    return res

这里的问题是，该proc.returncode值有时会变成None。我想，我对如何proc.wait()工作以及何时可以安全地停止读取输出有一个误解。如何使用 asyncio 子进程实现连续输出？

Answer 1

您的代码对我来说工作正常。您尝试运行的哪个命令导致了您的问题？

我能想到可以提供帮助的两件事是

1. 不要事后调用 .wait()，而是将返回码设置为循环条件以继续运行。
2. 不要等待整行返回，以防程序ffmpeg会执行一些技巧将其自身粘贴到控制台中，而不是实际发送换行符。

示例代码：

import asyncio, shlex, subprocess, sys


async def subprocess_async(cmd, **kwargs):
    cmd_list = shlex.split(cmd)
    proc = await asyncio.create_subprocess_exec(
        *cmd_list,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
        **kwargs)

    full_log = b""
    while proc.returncode is None:
        buf = await proc.stdout.read(20)
        if not buf:
            break
        full_log += buf
        sys.stdout.write(buf.decode())
    res = subprocess.CompletedProcess(cmd, proc.returncode,  stdout=full_log, stderr=b'')
    return res


if __name__ == '__main__':
    asyncio.run(subprocess_async("ffprobe -i video.mp4"))