Python 子流程模块是一个功能强大的库，用于启动和与子流程交互。它附带了一些高级 api，比如调用、检查输出和(从 Python 3.7开始)运行，这些都集中在的程序运行和等待完成的子进程上。

在这篇文章中，想讨论这个任务的一个变体，它不太直接涉及一个长时间运行的子进程。考虑测试一些服务器——例如 HTTP 服务器。将它作为一个子进程启动，然后将客户机连接到它，并运行一些测试序列。当完成后，希望以一种有序的方式关闭儿童程序。这对于同步运行子进程的 api 来说是很难实现的，因此必须查看一些较低级别的 api。

最近做的项目安卓 shell 里面有些就需要这个。

当然，可以使用 subprocess.run在一个线程中启动一个子进程，并在另一个线程中与其交互(例如，通过一个已知的端口， 比如HTTPServer)。但是，当完成了子进程之后，要完全终止它将变得非常棘手。如果子进程有一个有序的终止序列(比如发送某种“退出”命令) ，那么这是可行的。但是大多数服务器不这样做，只会永远旋转直到死亡。这就是这篇文章所讨论的用例。

启动，交互，终止，并获得所有输出时完成

第一个最简单的用例是启动一个 HTTP 服务器，与它交互，干净利落地终止它，并在完成后获取所有服务器的 stdout 和 stderr。下面是这些代码的重要部分(本文的所有代码示例都可以在这里找到) ，用 Python 3.6进行了测试:

defmain():proc=subprocess.Popen(['python3','-u','-m','http.server','8070'],stdout=subprocess.PIPE,stderr=subprocess.STDOUT)try:time.sleep(0.2)resp=urllib.request.urlopen('http://localhost:8070')assertb'Directory listing'inresp.read()finally:proc.terminate()try:outs,_=proc.communicate(timeout=0.2)print('== subprocess exited with rc =',proc.returncode)print(outs.decode('utf-8'))exceptsubprocess.TimeoutExpired:print('subprocess did not terminate in time')

子进程是一个 HTTP 服务器，使用 Python 自己的HTTP.server 模块，从启动它的目录中提供内容。使用底层的 Popen API 异步启动进程(意味着 Popen 立即返回，子进程在后台运行)。

请注意在调用时传递给 Python 的 -u: 这对于避免标准输出缓冲并在进程被终止时尽可能多地查看标准输出非常关键。在与子进程交互时，缓冲是一个严重的问题，稍后将看到更多这方面的示例。

样品的肉发生在最后一块。terminate()向子进程发送一个 SIGTERM 信号。然后，proc.communicate等待孩子退出并捕获所有的标准输出。Communicate 有一个非常方便的超时参数，从 Python 3.3[1]开始，让知道子进程是否由于某种原因没有退出。一个更复杂的技术是，如果子程序由于 SIGTERM 而没有退出，那么可以给子程序发送一个 SIGKILL (带有 proc.kill)。

如果你运行这个脚本，你会看到输出:

==subprocess exitedwithrc=-15ServingHTTP on0.0.0.0port8070(http://0.0.0.0:8070/) ...127.0.0.1--[10/Jun/202000:04:47]"GET / HTTP/1.1"200-

子进程的返回码是 -15(负的意思是以信号结束，15是 SIGTERM 的数字代码)。标准输出被正确地捕获并打印出来。

启动，交互，实时输出，终止

一个相关的用例是以“实时”方式获取子进程的标准输出，而不是在最后将所有内容放在一起。在这里，必须非常小心缓冲，因为它很容易导致程序崩溃和死锁。Linux 进程通常在交互模式下进行行缓冲，否则进行全缓冲。很少有进程是完全不缓冲的。因此，在看来，不建议在小于一行的块中读取 stdout。真的，千万别这么做。标准 i/o 意味着可以按行使用(想想所有的 Unix 命令行工具是如何工作的) ; 如果需要子行粒度，stdout 不是正确的方法(使用套接字或其他方法)。

无论如何，举个例子:

defoutput_reader(proc):forlineiniter(proc.stdout.readline,b''):print('got line: {0}'.format(line.decode('utf-8')),end='')defmain():proc=subprocess.Popen(['python3','-u','-m','http.server','8070'],stdout=subprocess.PIPE,stderr=subprocess.STDOUT)t=threading.Thread(target=output_reader,args=(proc,))t.start()try:time.sleep(0.2)foriinrange(4):resp=urllib.request.urlopen('http://localhost:8070')assertb'Directory listing'inresp.read()time.sleep(0.1)finally:proc.terminate()try:proc.wait(timeout=0.2)print('== subprocess exited with rc =',proc.returncode)exceptsubprocess.TimeoutExpired:print('subprocess did not terminate in time')t.join()

这个示例与之类似，除了标准输出的处理方式; 不再需要通信调用; 相反，proc.wait 只是等待子级退出(在发送 SIGTERM 之后)。线程会轮询子标准输出属性，只要有新行可用，就会循环并立即打印它们。如果运行这个示例，您将注意到孩子的 stdout 是实时报告的，而不是在最后报告一个错误。

(proc.stdout. readline，b”)代码片段继续调用 proc.stdout.readline() ，直到这个调用返回一个空的字节串。只有当关闭 proc.stdout 时才会发生这种情况，这种情况发生在子节点退出时。因此，尽管看起来读线程可能永远不会终止——但它总会终止！只要子进程在运行，线程就会忠实地阻塞该 readline; 只要子进程终止，readline 调用返回 b” ，线程就会退出。

如果不想仅仅打印捕获的 stdout，而是要对其进行处理(比如寻找预期的模式) ，那么可以使用 Python 的线程安全队列进行组织。读者的思路是:

defoutput_reader(proc,outq):forlineiniter(proc.stdout.readline,b''):outq.put(line.decode('utf-8'))

用以下方式启动它:

outq=queue.Queue()t=threading.Thread(target=output_reader,args=(proc,outq))t.start()

然后在任何时候，都可以使用非阻塞模式检查队列中是否有东西(完整代码示例在这里) :

try:line=outq.get(block=False)print('got line from outq: {0}'.format(line),end='')exceptqueue.Empty:print('could not get line from queue')

与子标准输入和标准输出的直接交互

这个示例进入了危险的水域; 子流程模块文档警告不要执行这里描述的事情，因为可能会出现死锁，但有时候根本就没有选择！有些程序喜欢使用它们的标准输入和标准输出进行交互。或者，您可能有一个具有交互(解释器)模式的程序，您希望对它进行测试——类似于 Python interepreter 本身。有时候可以一次提供这个程序的所有输入，然后检查它的输出; 这可以，也应该通过 communicate 来完成——这是完美的 API。它正确地输入 stdin，完成后关闭它(这意味着许多交互式程序游戏结束) ，等等。但是，如果真的希望基于子进程以前的一些输出提供额外的输入，该怎么办呢。下面是:

defmain():proc=subprocess.Popen(['python3','-i'],stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE)# To avoid deadlocks: careful to: add \n to output, flush output, use# readline() rather than read()proc.stdin.write(b'2+2\n')## 当将输入发送到行解释器时，不要忘记发送实际的换行符proc.stdin.flush()## 将数据放入流后，始终刷新流，因为它可能会被缓冲print(proc.stdout.readline())## 从行解释器获取输入proc.stdin.write(b'len("foobar")\n')proc.stdin.flush()print(proc.stdout.readline())proc.stdin.close()proc.terminate()proc.wait(timeout=0.2)

将数据发送到子标准输入，但由于某些原因(缺少换行、缓冲等) ，它无法获得完整的输入 然后调用 readline 等待回复，因为子进程仍然在等待输入完成(步骤1) ，所以的步骤2可能会永远挂起。这是典型的僵局。

在交互的最后，关闭子进程的 stdin (这是可选的，但对于某些类型的子进程很有用) ，调用 terminate，然后等待。最好是向子进程发送某种类型的“ exit”命令(对于 Python 解释器而言是 quit()) ; 这里的 terminate 是为了演示在其他选项不可用时必须做什么。注意，也可以在这里使用communicate，而不是等待来捕获 stderr 输出。

使用非阻塞读线程和可阻塞线程进行交互

最后的示例演示了一个稍微更高级的场景。假设正在测试一个长期存在的套接字服务器，并且有兴趣编排与它的复杂交互，可能是与多个并发客户机进行交互。还希望彻底关闭线程和子进程的整个设置。完整的代码示例在这里; 下面是一些代表性的代码片段。关键的部分是这个插座读取功能，意味着在它自己的线程中运行:

defsocket_reader(sockobj,outq,exit_event):whilenotexit_event.is_set():try:buf=sockobj.recv(1)iflen(buf)<1:breakoutq.put(buf)exceptsocket.timeout:continueexceptOSErrorase:break

最好与设置了超时的套接字一起使用，这个函数将重复监视套接字以获取新数据，并将接收到的所有数据推送到 outq (一个队列)中。排队。当套接字关闭(recv 返回一个空的字节串)或退出事件(一个线程)时，函数退出。事件)由调用方设置。

调用者可以在线程中启动这个函数，偶尔尝试以非阻塞的方式从队列中读取新项目

try:v=outq.get(block=False)print(v)exceptqueue.Empty:break

完成所有操作后，调用方可以设置 exit Event 来停止线程(如果线程读取的套接字已关闭，那么线程将自动停止，但是该事件允许更直接地控制这个操作)。对于本文中描述的任务，没有一个万能的解决方案; 展示了一系列处理更常见情况的处理方法，但是特定的用例可能不会被它们处理。