使用进程池/线程池加速 Python数据处理

目标
- 简单模式
- 多进程模式
参考

Python 是一种出色的编程语言，可用于处理数据和自动执行重复任务。尽管 Python 使编码变得有趣，但它并不总是运行速度最快的。默认情况下，Python 程序使用单个 CPU 作为单个进程执行。

如果有一台近十年生产的计算机，它很可能有 4 个（或更多）CPU 核心。这意味着在等待程序完成运行时，计算机 75% 或更多的资源几乎处于闲置状态！如何通过并行运行 Python 函数来充分利用计算机的处理能力。得益于Python的concurrent.futures模块，只需要3行代码就可以将一个普通程序变成可以并行处理数据的程序。

目标

假设有一个装满照片的文件夹，想要创建每张照片的缩略图。这是一个简短的程序，它使用 Python 的内置 glob 函数获取文件夹中所有 jpeg 文件的列表，然后使用 Pillow 图像处理库保存每张照片的 128 像素缩略图；

简单模式

# thumbnails_1.py
import glob
import os
from PIL import Imagedef make_image_thumbnail(filename):# 缩略图将被命名为"<original_filename>_thumbnail.jpg"base_filename, file_extension = os.path.splitext(filename)thumbnail_filename = f"{base_filename}_thumbnail{file_extension}"# 创建和存储缩略图image = Image.open(filename)image.thumbnail(size=(128, 128))image.save(thumbnail_filename, "JPEG")return thumbnail_filename# 遍历文件夹下的所有jpg文件并为每一张图生成缩略图
for image_file in glob.glob("*.jpg"):thumbnail_file = make_image_thumbnail(image_file)print(f"A thumbnail for {image_file} was saved as {thumbnail_file}")

该程序运行时间为 8.9 秒。但是计算机大概有75%的cpu处于空闲状态；问题是电脑有 4 个 CPU 核心，但 Python 只使用其中之一。因此当最大限度地发挥一个 CPU时，其他三个 CPU 却没有执行任何操作。

多进程模式

将 jpeg 文件列表分成 4 个较小的块。运行 4 个独立的 Python 解释器实例。让每个 Python 实例处理 4 个数据块之一。合并 4 个过程的结果即可得到最终结果列表。在四个独立的 CPU 上运行的四个 Python 副本应该能够完成大约是一个 CPU 的 4 倍的工作量，对吗？
只需要3步：

导入concurrent.futures库；
启动 4个Python 实例通过创建一个进程池来做到这一点；
要求进程池使用这 4 个进程在数据列表上执行辅助函数。executor.map() 函数接受要调用的辅助函数以及要使用它处理的数据列表。它完成了拆分列表、将子列表发送到每个子进程、运行子进程以及组合结果等所有艰苦工作。非常的简洁！

executor.map() 函数返回结果的顺序与给它处理的数据列表的顺序相同。所以使用Python的zip()函数作为快捷方式来一步获取原始文件名和匹配结果。

# thumbnails_2.pyimport glob
import os
from PIL import Image
import concurrent.futuresdef make_image_thumbnail(filename):# 缩略图将被命名为"<original_filename>_thumbnail.jpg"base_filename, file_extension = os.path.splitext(filename)thumbnail_filename = f"{base_filename}_thumbnail{file_extension}"# 创建和存储缩略图image = Image.open(filename)image.thumbnail(size=(128, 128))image.save(thumbnail_filename, "JPEG")return thumbnail_filename# 创建一个线程池处理，为每个cpu创建一个实例
with concurrent.futures.ProcessPoolExecutor() as executor:# 获取所有要处理的jpg文件image_files = glob.glob("*.jpg")# 处理文件列表 将任务拆分到线程池以利用所有的cpufor image_file, thumbnail_file in zip(image_files, executor.map(make_image_thumbnail, image_files)):print(f"A thumbnail for {image_file} was saved as {thumbnail_file}")

2.2秒就完成了！与原始版本相比，速度提高了 4 倍。由于使用 4 个 CPU 而不是 1 个，因此运行时间更快。但如果仔细观察，你会发现“用户”时间几乎是 9 秒。程序如何在 2.2 秒内完成但仍然运行了 9 秒？这似乎……不可能？这是因为“用户”时间是所有 CPU 的 CPU 时间的总和。

生成更多 Python 进程并在它们之间调整数据会产生一些开销，因此并不总是能获得如此大的速度提升。 如果正在处理巨大的数据集，那么设置 chunksize 参数的技巧可以提供很大帮助。当有要处理的数据列表并且每条数据都可以独立处理时，使用进程池是一个很好的解决方案。

以下是适合多处理的一些示例：

从一组单独的 Web 服务器日志文件中获取统计信息
从一堆 XML、CSV 或 json 文件中解析数据
预处理大量图像以创建机器学习数据集

但进程池并不总是答案。使用进程池需要在单独的 Python 进程之间来回传递数据。如果正在使用的数据无法在进程之间有效传递，那么这将不起作用。 如果你需要处理上一条数据的结果来处理下一条数据，这也是行不通的。这种情况下适合用Python 有一个全局解释器锁（Global Interpreter Lock），即 GIL。这意味着即使程序是多线程的，任何线程一次也只能执行一条 Python 代码指令。换句话说，多线程Python代码无法真正并行运行。但进程池可以解决这个问题！ 因为运行的是真正独立的 Python 实例，所以每个实例都有自己的 GIL。

可以获得 Python 代码的真正并行执行（以一些额外开销为代价）。不要害怕并行处理！借助concurrent.futures 库，Python提供了一种简单的方法来调整脚本以同时使用计算机中的所有 CPU 核心。