《asyncio 系列》13. 在 asyncio 中调用命令行程序

楔子

Python 编写的应用程序需要 Python 运行环境，然而，并不是所有我们想要交互的组件都是用 Python 编写的。我们可能已经使用了一个用 C++ 、Go、Rust 或其他语言编写的应用程序，这些语言提供了更好的运行时特征，或者可以提供很优秀的实现方式，而无需我们重新实现。可能还希望使用操作系统提供的命令行实用工具，例如 grep 用于搜索大型文件，curl 用于发出 HTTP 请求等。

在标准 Python 中，可使用 subprocess 模块在单独的进程中运行不同的应用程序。与大多数其他 Python模块一样，标准子进程 API 是阻塞的，这使得它在没有多线程或 multiprocessing 的情况下与 asyncio 不兼容。为此 asyncio 提供了一个以 subprocess 模块为原型的模块，从而使用协程异步创建和管理子进程。

创建子进程

假设你想扩展现有 Python Web API 的功能，组织中的另一个团队已经为他们拥有的批处理机制在命令行应用程序中构建了你想要的功能，但主要问题是该应用程序是用 Rust 编写的。鉴于应用程序已经存在，你不希望在 Python 中重新实现，那有没有办法让我们仍然可以在现有的 Python API 中使用这个应用程序的功能呢？

我们可使用子进程来重用这个应用程序，然后读取应用程序的结果，并根据需要在现有 API 中使用它，从而省去了重新实现应用程序的麻烦。

asyncio 提供了两个开箱即用的协程函数来创建子进程：asyncio.create_subprocess_shell 和 asyncio.create_subprocess_exec。这些协程函数都返回一个 Process 实例，它具有让我们等待进程完成和终止进程的能力。那问题来了，为什么要用两个协程来完成看似相同的任务呢？首先 create_subprocess_shell 协程函数在操作系统的 shell 中创建一个子进程，例如 zsh 或 bash，一般来说，除非你需要使用 shell 的功能，否则最好使用 create_subprocess_exec。使用 shell 可能会有一些陷阱，例如不同的机器使用不同的 shell，或者相同的 shell 配置不同。这很难保证应用程序在不同的机器上具有相同的表现。

要了解创建子进程的基础知识，让我们编写一个异步应用程序来运行一个简单的命令行程序。将从 ls 程序开始，它可以列出当前目录中的内容（尽管我们不太可能在现实世界中这样做）。如果在 Windows 机器上运行，请将 Is -l 替换为 dir。

import asyncio
from asyncio.subprocess import Process

async def main():
    process: Process = await asyncio.create_subprocess_exec("ls", "-l")
    print(f"进程的 pid: {process.pid}")
    # 等待子进程执行完毕，并返回状态码
    status_code = await process.wait()
    print(f"status code: {status_code}")

asyncio.run(main())

执行之后会列出当前目录下的所有文件：

在代码中我们使用 create_subprocess_exec 运行 ls 命令，并返回一个 Process 实例。还可通过在后面添加其他参数来指定传递给程序的参数，这里传入 -l 添加一些关于文件的额外信息。创建进程后，输出进程 ID，然后调用 wait 协程，这个协程会一直等到子进程运行完毕，一旦完成就会返回子进程的状态码。这种情况下它应该为零，这里没有截出来。默认情况下，子进程的标准输出将通过管道传输到我们自己应用程序的标准输出。当运行它时，你应该会看到 ls -l 的运行输出。

注意：wait 协程将阻塞，直到应用程序终止，并且无法保证进程需要多长时间才能终止，以及它是否会终止。如果你担心进程失控，则需要通过用 asyncio.wait_for 引入超时。然而这里有一个坑，回顾一下，wait_for 在超时的时候将终止正在运行的协程。你可能认为这将终止进程，但事实并非如此，它只终止等待进程完成的任务，而不终止底层进程。

我们需要一种更好的方法来在超时的时候关闭进程，幸运的是，Process 有两种方法可以帮助我们解决这个问题：terminate 和 kill。terminate 方法将向子进程发送 SIGTERM 信号，而 kill 将发送 SIGKILL 信号。请注意，这两种方法都不是协程，并且是非阻塞的。它们只是发送信号。如果你想在终止子进程后尝试获取返回码，或者想等待任何清理动作，则需要再次调用 wait。

编写代码时，要注意的一件事是 except 块内的 wait 仍然有可能需要很长时间，如果出现这种情况，需要再次将其包装在 wait_for 中。

控制标准输出

在前面的示例中，子进程的标准输出直接进入应用程序的标准输出。如果不想要这种行为怎么办？也许我们想对输出做额外处理，或者输出是无关紧要的（我们可放心地忽略它）。create_subprocess_exec 协程有一个 stdout 参数，可让我们指定希望输出到哪里。

假设我们计划同时运行多个子进程，并生成输出，但我们想知道哪个子进程生成了哪些输出，以避免混淆。为此，需要做的第一件事是将 stdout 参数设置为asyncio.subprocess.PIPE。这告诉子进程创建一个新的 StreamReader 实例，可使用它来读取进程的输出，然后可使用 Process.stdout 字段访问这个流读取器。

import asyncio
from asyncio.subprocess import Process, PIPE
from asyncio.streams import StreamReader

async def main():
    process: Process = await asyncio.create_subprocess_exec(
        "ls", "-la", stdout=PIPE)
    print(f"进程的 pid: {process.pid}")
    await process.wait()
    # 当子进程执行完毕时，拿到它的 stdout 属性
    stdout: StreamReader = process.stdout
    # 读取输出内容，如果子进程没有执行完毕，那么 await stdout.read() 会阻塞
    content = (await stdout.read()).decode("utf-8")
    print(content[: 100])


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

看一下输出：

结果正常，这里我们只打印了输出的前 100 个字节。然后使用管道以及处理子进程输入、输出的一个关键方面是它们容易出现死锁，如果子进程生成大量输出，而我们没有正确使用它，那么 wait 协程特别容易受到这种影响。为证明这一点，让我们看一个简单示例，它调用一个 Python 应用程序，该应用程序将大量数据写入标准输出，并一次性刷新所有数据。

# 文件名：write_data.py
import sys
[sys.stdout.buffer.write(b'!Hello there!!\n') for _ in range(1000000)]
sys.stdout.flush()

# 文件名：main.py
import asyncio
from asyncio.subprocess import Process, PIPE

async def main():
    process: Process = await asyncio.create_subprocess_exec(
        "python", "write_data.py", stdout=PIPE)
    print(f"进程的 pid: {process.pid}")
    await process.wait()
    # await process.wait() 会返回执行后的状态码
    # 也可以直接通过 process.returncode 获取（如果子进程没有执行完，则结果为 None）
    print(f"状态码: {process.returncode}")

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

如果运行代码，你将看到进程 pid 被输出，但该应用程序将永远挂起，你需要强制终止它。如果你在运行该代码时，没有看到这样的结果，只需要增加应用程序中输出数据的次数，你终将遇到这个问题。

应用程序看起来很简单，那为什么会遇到这种死锁呢？问题在于流读取器的缓冲区是如何工作的。当流读取器的缓冲区被填满时，写入它的其他任何调用都会被阻塞，直到缓冲区中有更多可用空间。虽然流读取器缓冲区因缓冲区已满而被阻塞，但进程仍在尝试将其大量输出写入流读取器。这使得进程可依赖于流读取器，但流读取器永远不会正常运行，因为我们永远不会释放缓冲区中的任何空间，这种循环依赖造成了死锁。

可以通过避免使用 wait 协程来解决此问题，Process 类有一个协程方法，称为通信（communicate），可以完全避免死锁。这个协程一直阻塞，直到子进程完成，并同时使用标准输出和标准错误，一旦应用程序完成就返回完整输出。让我们修改之前的示例，使用通信来解决问题。

import asyncio
from asyncio.subprocess import Process, PIPE

async def main():
    process: Process = await asyncio.create_subprocess_exec(
        "python", "write_data.py", stdout=PIPE, stderr=PIPE)
    print(f"进程的 pid: {process.pid}")
    # 同样会阻塞，直到进程完成
    stdout, stderr = await process.communicate()
    print(f"状态码: {process.returncode}")

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

未完待续