使用 asyncio 等待子进程的结果

我的 Python 脚本包含一个循环，该循环使用subprocess在脚本之外运行命令。每个子进程都是独立的。我监听返回的消息，以防出现错误；我不能忽略子进程的结果。这是没有 asyncio 的脚本（我已将计算量大的调用替换为sleep):

from subprocess import PIPE  # https://docs.python.org/3/library/subprocess.html
import subprocess

def go_do_something(index: int) -> None:
    """
    This function takes a long time
    Nothing is returned
    Each instance is independent
    """
    process = subprocess.run(["sleep","2"],stdout=PIPE,stderr=PIPE,timeout=20)
    stdout = process.stdout.decode("utf-8")
    stderr = process.stderr.decode("utf-8")
    if "error" in stderr:
        print("error for "+str(index))
    return

def my_long_func(val: int) -> None:
    """
    This function contains a loop
    Each iteration of the loop calls a function
    Nothing is returned
    """
    for index in range(val):
        print("index = "+str(index))
        go_do_something(index)

# run the script
my_long_func(3) # launch three tasks

我想我可以用asyncio加速此活动，因为 Python 脚本正在等待外部subprocess去完成。我认为threading or multiprocessing不是必需的，尽管它们也可以提高执行速度。使用任务队列（例如 Celery）是另一种选择。

我尝试实施asyncio方法，但缺少一些东西，因为以下尝试不会改变总体执行时间：

import asyncio
from subprocess import PIPE  # https://docs.python.org/3/library/subprocess.html
import subprocess


async def go_do_something(index: int) -> None:
    """
    This function takes a long time
    Nothing is returned
    Each instance is independent
    """
    process = subprocess.run(["sleep","2"],stdout=PIPE,stderr=PIPE,timeout=20)
    stdout = process.stdout.decode("utf-8")
    stderr = process.stderr.decode("utf-8")
    if "error" in stderr:
        print("error for "+str(index))
    return

def my_long_func(val: int) -> None:
    """
    This function contains a loop
    Each iteration of the loop calls a function
    Nothing is returned
    """
    # https://docs.python.org/3/library/asyncio-eventloop.html
    loop = asyncio.get_event_loop()
    tasks = []
    for index in range(val):
        task = go_do_something(index)
        tasks.append(task)
    # https://docs.python.org/3/library/asyncio-task.html
    tasks = asyncio.gather(*tasks)
    loop.run_until_complete(tasks)
    loop.close()
    return

my_long_func(3) # launch three tasks

如果我想监控每个的输出subprocess但不要等待subprocess跑步，我能从中受益吗asyncio？或者这种情况是否需要类似的东西multiprocessing还是芹菜？

尝试使用执行命令asyncio代替subprocess.

定义一个run()功能：

import asyncio

async def run(cmd: str):
    proc = await asyncio.create_subprocess_shell(
        cmd,
        stderr=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE
    )

    stdout, stderr = await proc.communicate()

    print(f'[{cmd!r} exited with {proc.returncode}]')
    if stdout:
        print(f'[stdout]\n{stdout.decode()}')
    if stderr:
        print(f'[stderr]\n{stderr.decode()}')

然后你可以像调用任何函数一样调用它async功能：

asyncio.run(run('sleep 2'))

#=>

['sleep 2' exited with 0]

例子取自官方文档。也提供here.