MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Python 线程替代方案之 subprocess 模块

2022-01-132.0k 阅读

Python 线程的局限性

在 Python 编程中，线程（Threading）是一种常用的并发处理方式。然而，Python 的线程存在一些局限性，这使得在某些场景下需要寻找替代方案。

GIL 限制

Python 有一个全局解释器锁（Global Interpreter Lock，GIL）。GIL 是一个互斥锁，它确保在任何时刻只有一个线程可以执行 Python 字节码。这意味着，即使在多核处理器上，Python 的多线程程序也无法真正利用多核的优势来并行执行 Python 代码。例如，在进行 CPU 密集型任务时，多线程程序可能无法获得预期的性能提升，因为线程之间会频繁地竞争 GIL，导致大部分时间都在等待锁的释放。

import threading
import time


def cpu_bound_task():
    result = 0
    for i in range(100000000):
        result += i
    return result


start_time = time.time()
threads = []
for _ in range(4):
    t = threading.Thread(target=cpu_bound_task)
    threads.append(t)
    t.start()

for t in threads:
    t.join()
end_time = time.time()
print(f"Total time with threads: {end_time - start_time} seconds")

在上述代码中，我们创建了 4 个线程来执行 CPU 密集型任务。但由于 GIL 的存在，实际执行时间可能并不会比单线程快很多，甚至可能更慢，因为线程切换和 GIL 竞争带来了额外的开销。

线程安全问题

多线程编程需要处理线程安全问题。当多个线程同时访问和修改共享资源时，可能会导致数据竞争和不一致的结果。例如，多个线程同时对一个全局变量进行加 1 操作，如果没有适当的同步机制（如锁），最终的结果可能是错误的。

import threading

counter = 0


def increment():
    global counter
    for _ in range(100000):
        counter += 1


threads = []
for _ in range(10):
    t = threading.Thread(target=increment)
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"Final counter value: {counter}")

在这个例子中，由于多个线程同时修改 counter 变量，最终的结果可能不是预期的 1000000（10 * 100000），因为线程之间的数据竞争导致了结果的不确定性。

subprocess 模块简介

subprocess 模块是 Python 标准库中用于创建新进程、连接到它们的输入/输出/错误管道以及获取它们的返回码的模块。它提供了一种强大的方式来与外部程序进行交互，并且可以作为线程的替代方案在某些场景下使用。

创建子进程

subprocess 模块中最基本的函数是 subprocess.run()。这个函数用于执行一个外部命令，并等待命令完成。它的基本语法如下：

subprocess.run(args, *, stdin=None, input=None, stdout=None, stderr=None, shell=False, timeout=None, check=False)
  • args：要执行的命令及其参数，可以是字符串或字符串序列。
  • stdin、stdout、stderr：分别用于指定标准输入、标准输出和标准错误的处理方式，可以是 subprocess.PIPE（用于捕获输出）、subprocess.DEVNULL（用于丢弃输出）或文件对象。
  • input：如果指定了 input，则会将其作为标准输入传递给子进程。
  • shell：如果设置为 True，则会通过 shell 来执行命令。一般情况下，建议设置为 False，以避免潜在的安全风险。
  • timeout：设置子进程执行的超时时间，如果超过这个时间子进程仍未完成，会引发 TimeoutExpired 异常。
  • check：如果设置为 True，当子进程返回非零返回码时，会引发 CalledProcessError 异常。

以下是一个简单的示例，使用 subprocess.run() 执行 ls 命令，并捕获其输出：

import subprocess

result = subprocess.run(['ls', '-l'], stdout=subprocess.PIPE, text=True)
print(result.stdout)

在上述代码中，我们使用 subprocess.run() 执行了 ls -l 命令，并通过 stdout=subprocess.PIPE 捕获了命令的输出。text=True 用于将输出以文本形式返回，而不是字节形式。

异步执行子进程

除了 subprocess.run() 之外，subprocess 模块还提供了 subprocess.Popen 类，用于异步执行子进程。Popen 类允许我们在启动子进程后继续执行主程序，而不需要等待子进程完成。

import subprocess
import time

# 异步启动一个子进程
process = subprocess.Popen(['ping', 'www.google.com'], stdout=subprocess.PIPE, text=True)

# 主程序继续执行其他任务
while True:
    output = process.stdout.readline()
    if output == '' and process.poll() is not None:
        break
    if output:
        print(output.strip())
    time.sleep(1)

# 获取子进程的返回码
return_code = process.poll()
print(f"Subprocess return code: {return_code}")

在这个例子中，我们使用 subprocess.Popen 启动了一个 ping 命令的子进程。主程序在启动子进程后，通过循环读取子进程的输出，并在子进程完成后获取其返回码。

subprocess 作为线程替代方案的应用场景

CPU 密集型任务

如前文所述，Python 线程在处理 CPU 密集型任务时由于 GIL 的限制无法充分利用多核优势。而使用 subprocess 模块创建多个子进程来处理 CPU 密集型任务，可以绕过 GIL 的限制，充分利用多核处理器的性能。

import subprocess
import time


def cpu_bound_task():
    subprocess.run(['python', '-c', 'for i in range(100000000): pass'])


start_time = time.time()
processes = []
for _ in range(4):
    p = subprocess.Popen(['python', '-c', 'for i in range(100000000): pass'])
    processes.append(p)

for p in processes:
    p.wait()
end_time = time.time()
print(f"Total time with subprocess: {end_time - start_time} seconds")

在这个示例中，我们创建了 4 个子进程，每个子进程都执行一个简单的 CPU 密集型任务。由于每个子进程都是独立的 Python 进程，不受 GIL 的限制，因此可以充分利用多核处理器的性能，相比使用线程可能会获得更好的性能提升。

外部程序调用

在很多情况下，我们需要在 Python 程序中调用外部的可执行程序。例如，调用系统命令行工具、运行其他编程语言编写的程序等。使用 subprocess 模块可以方便地实现这一点，并且可以对外部程序的输入输出进行灵活的控制。

import subprocess

# 调用 ImageMagick 的 convert 命令来调整图片大小
subprocess.run(['convert', 'input.jpg', '-resize', '50%', 'output.jpg'])

在这个例子中，我们使用 subprocess.run() 调用了 ImageMagick 工具中的 convert 命令，将 input.jpg 图片调整为原来大小的 50% 并保存为 output.jpg。通过这种方式，我们可以将 Python 与各种外部工具集成，扩展 Python 程序的功能。

隔离和安全性

使用 subprocess 创建的子进程与主 Python 进程是相互隔离的。这意味着子进程中的错误不会影响主进程的稳定性，并且子进程可以在有限的权限下运行，提高了程序的安全性。例如，我们可以在子进程中运行一些可能存在风险的代码，而不会对主进程造成威胁。

import subprocess

try:
    subprocess.run(['python', 'risky_script.py'], check=True)
except subprocess.CalledProcessError as e:
    print(f"Subprocess failed with return code {e.returncode}")

在这个示例中，risky_script.py 可能是一个包含潜在风险操作的 Python 脚本。通过 subprocess.run() 运行它，并使用 check=True 来捕获子进程的返回码，如果子进程出现错误（返回非零返回码），主程序可以捕获异常并进行相应的处理，而不会导致主程序崩溃。

subprocess 模块的高级用法

管道操作

subprocess 模块支持管道操作，即可以将一个子进程的输出作为另一个子进程的输入。这在处理复杂的命令链时非常有用。

import subprocess

# 查找当前目录下所有文件，并统计文件数量
process1 = subprocess.Popen(['ls', '-l'], stdout=subprocess.PIPE)
process2 = subprocess.Popen(['wc', '-l'], stdin=process1.stdout, stdout=subprocess.PIPE)
process1.stdout.close()  # 关闭 process1 的 stdout，避免资源泄漏
output, _ = process2.communicate()
print(f"Number of files: {output.decode('utf-8').strip()}")

在这个例子中，我们首先使用 ls -l 命令列出当前目录下的文件列表，并将其输出通过管道传递给 wc -l 命令，用于统计文件的数量。通过 subprocess.Popen 的组合使用，实现了类似 Unix 管道的功能。

与子进程交互

有时候我们需要与正在运行的子进程进行交互，例如向子进程发送输入、获取子进程的实时输出等。subprocess.Popen 类提供了 communicate() 方法来实现这一点。

import subprocess

# 启动一个交互式的 Python 解释器子进程
process = subprocess.Popen(['python'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)

# 向子进程发送命令并获取输出
output, error = process.communicate('print("Hello, subprocess!")\n')
print(f"Output: {output}")
print(f"Error: {error}")

在这个示例中，我们启动了一个交互式的 Python 解释器子进程，并通过 communicate() 方法向其发送了一条 Python 语句 print("Hello, subprocess!")，然后获取了子进程的输出和错误信息。

性能比较：线程 vs subprocess

为了更直观地了解线程和 subprocess 在不同场景下的性能差异，我们进行一些简单的性能测试。

CPU 密集型任务性能测试

import threading
import subprocess
import time


def cpu_bound_thread_task():
    result = 0
    for i in range(100000000):
        result += i
    return result


def cpu_bound_subprocess_task():
    subprocess.run(['python', '-c', 'for i in range(100000000): pass'])


# 线程性能测试
start_time_thread = time.time()
threads = []
for _ in range(4):
    t = threading.Thread(target=cpu_bound_thread_task)
    threads.append(t)
    t.start()

for t in threads:
    t.join()
end_time_thread = time.time()
print(f"Total time with threads: {end_time_thread - start_time_thread} seconds")

# subprocess 性能测试
start_time_subprocess = time.time()
processes = []
for _ in range(4):
    p = subprocess.Popen(['python', '-c', 'for i in range(100000000): pass'])
    processes.append(p)

for p in processes:
    p.wait()
end_time_subprocess = time.time()
print(f"Total time with subprocess: {end_time_subprocess - start_time_subprocess} seconds")

在这个性能测试中，我们分别使用线程和 subprocess 来执行 4 个 CPU 密集型任务。通常情况下，由于 GIL 的限制，线程版本的执行时间会比 subprocess 版本长，尤其是在多核处理器上，subprocess 可以更好地利用多核性能。

I/O 密集型任务性能测试

import threading
import subprocess
import time


def io_bound_thread_task():
    with open('test.txt', 'r') as f:
        data = f.read()
    return data


def io_bound_subprocess_task():
    subprocess.run(['cat', 'test.txt'])


# 线程性能测试
start_time_thread = time.time()
threads = []
for _ in range(10):
    t = threading.Thread(target=io_bound_thread_task)
    threads.append(t)
    t.start()

for t in threads:
    t.join()
end_time_thread = time.time()
print(f"Total time with threads: {end_time_thread - start_time_thread} seconds")

# subprocess 性能测试
start_time_subprocess = time.time()
processes = []
for _ in range(10):
    p = subprocess.Popen(['cat', 'test.txt'])
    processes.append(p)

for p in processes:
    p.wait()
end_time_subprocess = time.time()
print(f"Total time with subprocess: {end_time_subprocess - start_time_subprocess} seconds")

在 I/O 密集型任务中，线程和 subprocess 的性能差异可能没有 CPU 密集型任务那么明显。由于 I/O 操作通常会释放 GIL，线程在这种情况下也能有较好的表现。然而，subprocess 创建和管理进程的开销相对较大，所以在 I/O 密集型任务中，线程可能会更具优势，具体性能还取决于系统资源和任务的具体性质。

注意事项和潜在问题

资源消耗

创建子进程会消耗更多的系统资源，包括内存和 CPU 时间。每个子进程都有自己独立的地址空间和资源，因此在使用 subprocess 时需要谨慎考虑系统的承载能力，避免创建过多的子进程导致系统资源耗尽。

错误处理

在使用 subprocess 时，需要妥善处理子进程可能出现的各种错误。例如，子进程可能因为命令不存在、参数错误等原因无法正常执行，此时需要通过捕获异常或检查返回码来进行相应的处理，以确保程序的稳定性和可靠性。

跨平台兼容性

subprocess 模块在不同的操作系统上可能有一些细微的差异。例如，某些系统命令在不同操作系统上的名称和参数可能不同。在编写跨平台的程序时，需要注意这些差异，并进行适当的兼容性处理。

通过深入了解 Python 线程的局限性以及 subprocess 模块的特性和用法，我们可以在不同的场景下选择合适的并发处理方式，提高 Python 程序的性能和稳定性。无论是处理 CPU 密集型任务、调用外部程序还是需要更高的隔离性和安全性，subprocess 模块都提供了一种强大的线程替代方案。在实际应用中，需要根据具体的需求和系统环境，合理地使用线程和 subprocess，以达到最佳的编程效果。同时，要注意处理好资源消耗、错误处理和跨平台兼容性等问题，确保程序的健壮性和可移植性。