第17章：使用 concurrent.futures 模块处理并发-阻塞型 IO 和 GIL、使用 ProcessPoolExecutor 启动进程

17.2 阻塞型I/O和GIL

CPython 解释器本身就不是线程安全的，因此有全局解释器锁(GIL)， 一次只允许使用一个线程执行 Python 字节码。因此，一个 Python 进程通常不能同时使用多个 CPU 核心。

编写 Python 代码时无法控制 GIL；不过，执行耗时的任务时，可以使用一个内置的函数或一个使用 C 语言编写的扩展释放 GIL。其实，有个使用 C 语言编写的 Python 库能管理 GIL，自行启动操作系统线程，利用全部可用的 CPU 核心。这样做会极大地增加库代码的复杂度，因此大多数库的作者都不这么做。

然而，标准库中所有执行阻塞型 I/O 操作的函数，在等待操作系统返回结果时都会释放 GIL，time.sleep() 也会释放 GIL。这意味着在 Python 语言这个层次上可以使用多线程，但只能在执行 I/O 密集型任务时从中受益：一个 Python 线程等待网络响应时，阻塞型 I/O 函数会释放 GIL，再运行一个线程。

http://c.biancheng.net/view/5537.html

17.3 使用 concurrent.futures 模块启动进程

concurrent.futures 模块的文档副标题是“Launching parallel tasks”(执行并行任务)。这个模块实现的是真正的并行计算，因为它使用 ProcessPoolExecutor 类把工作分配给多个 Python 进程处理。因此，如果需要做 CPU 密集型处理，使用这个模块能绕开 GIL，利用所有可用的 CPU 核心。

ProcessPoolExecutor 和 ThreadPoolExecutor 类都实现了通用的 Executor 接口，因此使用 concurrent.futures 模块能特别轻松地把基于线程的方案转成基于进程的方案，例如 17-1-2 中下载国旗的示例，只要把 ThreadPoolExecutor 改成 ProcessPoolExecutor 就行：

def batch_downloads(): """多进程下载""" # 启动进程程池 with futures.ProcessPoolExecutor() as executor: res = executor.map(download_flag, sorted(POP20_CC)) # 返回结果列表 return list(res)

对简单的用途来说，这两个实现 Executor 接口的类唯一值得注意的区别是，ThreadPoolExecutor.__init__ 方法需要 max_workers 参数，指定线程池中线程的数量。在 ProcessPoolExecutor 类中，那个参数是可选的，而且大多数情况下不使用——默认值是 os.cpu_count() 函数返回的 CPU 数量。这样处理说得通，因为对 CPU 密集型的处理来说，不可能要求使用超过 CPU 数量的进程。而对 I/O 密集型处理来说，可以在一个 ThreadPoolExecutor 实例中使用 10 个、100 个或 1000 个线程；最佳线程数取决于做的是什么事，以及可用内存有多少，因此要仔细测试才能找到最佳的线程数。

经过几次测试，我发现使用 ProcessPoolExecutor 实例下载 20 面国旗的时间增加到了 3.69 秒，而原来使用 ThreadPoolExecutor 的版本是 1.2 秒。主要原因可能是，我的电脑用的是四核八线程 CPU，因此限制只能有 8 个并发下载，而使用线程池的版本有 20 个工作的线程。因此 ProcessPoolExecutor 的价值体现在 CPU 密集型作业上。

下面我们通过计算斐波那契数列的示例来演示 ProcessPoolExecutor 在 CPU 密集型作业的优势：

import time def fib(x): """计算斐波那契数列""" st = time.time() a = 0 b = 1 n = 0 while n < 500000: a, b = b, a + b n += 1 et = time.time() - st print('{} use time -> {:.2f}s'.format(x, et)) return x

先看看 ThreadPoolExecutor 的计算时间：

from concurrent import futures tasks = [0, 1, 2, 3, 4, 5, 6, 7] with futures.ThreadPoolExecutor() as executor: executor.map(fib, tasks) # 0 use time -> 25.43s # 4 use time -> 25.86s # 7 use time -> 25.69s # 6 use time -> 25.76s # 1 use time -> 26.18s # 2 use time -> 26.20s # 5 use time -> 26.22s # 3 use time -> 26.33s

再看看 ProcessPoolExecutor 的表现：

from concurrent import futures tasks = [0, 1, 2, 3, 4, 5, 6, 7] with futures.ProcessPoolExecutor() as executor: executor.map(fib, tasks) # 结果输出： # 1 use time -> 7.18s # 6 use time -> 7.18s # 0 use time -> 7.18s # 4 use time -> 7.22s # 3 use time -> 7.23s # 5 use time -> 7.26s # 2 use time -> 7.28s # 7 use time -> 7.30s