Question

为了高效处理网络 I/O，需要使用并发，因为网络有很高的延迟，所以为了不浪费 CPU 周期去等待，最好在收到网络响应之前做些其他的事。

为了通过代码说明这一点，我写了三个示例程序，从网上下载 20 个国家的国旗图像。第一个示例程序 flags.py 是依序下载的：下载完一个图像，并将其保存在硬盘中之后，才请求下一个图像。另外两个脚本是并发下载的：几乎同时请求所有图像，每下载完一个文件就保存一个文件。flags_threadpool.py 脚本使用 concurrent.futures 模块，而 flags_asyncio.py 脚本使用 asyncio 包。

示例 17-1 是运行这三个脚本得到的结果，每个脚本都运行三次。我还在 YouTube 上发布了一个 73 秒的视频，让你观看这些脚本的运行情况，你会看到一个 OS X Finder 窗口，显示运行过程中保存的国旗图像文件。这些脚本从 flupy.org 下载图像，而这个网站架设在 CDN 之后，因此第一次运行时可能要等很久才能看到结果。示例 17-1 中显示的结果是运行几次之后收集的，因此 CDN 中已经有了缓存。

示例 17-1　运行 flags.py、flags_threadpool.py 和 flags_asyncio.py 脚本得到的结果

$ python3 flags.py
BD BR CD CN DE EG ET FR ID IN IR JP MX NG PH PK RU TR US VN ➊
20 flags downloaded in 7.26s ➋
$ python3 flags.py
BD BR CD CN DE EG ET FR ID IN IR JP MX NG PH PK RU TR US VN
20 flags downloaded in 7.20s
$ python3 flags.py
BD BR CD CN DE EG ET FR ID IN IR JP MX NG PH PK RU TR US VN
20 flags downloaded in 7.09s
$ python3 flags_threadpool.py
DE BD CN JP ID EG NG BR RU CD IR MX US PH FR PK VN IN ET TR
20 flags downloaded in 1.37s ➌
$ python3 flags_threadpool.py
EG BR FR IN BD JP DE RU PK PH CD MX ID US NG TR CN VN ET IR
20 flags downloaded in 1.60s
$ python3 flags_threadpool.py
BD DE EG CN ID RU IN VN ET MX FR CD NG US JP TR PK BR IR PH
20 flags downloaded in 1.22s
$ python3 flags_asyncio.py ➍
BD BR IN ID TR DE CN US IR PK PH FR RU NG VN ET MX EG JP CD
20 flags downloaded in 1.36s
$ python3 flags_asyncio.py
RU CN BR IN FR BD TR EG VN IR PH CD ET ID NG DE JP PK MX US
20 flags downloaded in 1.27s
$ python3 flags_asyncio.py
RU IN ID DE BR VN PK MX US IR ET EG NG BD FR CN JP PH CD TR ➎
20 flags downloaded in 1.42s

❶ 每次运行脚本后，首先显示下载过程中下载完毕的国家代码，最后显示一个消息，说明耗时。

❷ flags.py 脚本下载 20 个图像平均用时 7.18 秒。

❸ flags_threadpool.py 脚本平均用时 1.40 秒。

❹ flags_asyncio.py 脚本平均用时 1.35 秒。

❺ 注意国家代码的顺序：对并发下载的脚本来说，每次下载的顺序都不同。

两个并发下载的脚本之间性能差异不大，不过都比依序下载的脚本快 5 倍多。这只是一个特别小的任务，如果把下载的文件数量增加到几百个，并发下载的脚本能比依序下载的脚本快 20 倍或更多。

　在公网中测试 HTTP 并发客户端可能不小心变成拒绝服务（Denial-of-Service，DoS）攻击，或者有这么做的嫌疑。我们可以像示例 17-1 那样做，因为那三个脚本被硬编码，限制只发起 20 个请求。如果想大规模测试 HTTP 服务器，应该自己架设测试服务器。在本书的 GitHub 仓库中，17-futures/countries/README.rst 文件说明了如何在本地架设 Nginx 服务器。

下面我们来分析示例 17-1 测试的两个脚本——flags.py 和 flags_threadpool.py，看看它们的实现方式。第三个脚本 flags_asyncio.py 留到第 18 章再分析。将这三个脚本一起演示是为了表明一个观点：在 I/O 密集型应用中，如果代码写得正确，那么不管使用哪种并发策略（使用线程或 asyncio 包），吞吐量都比依序执行的代码高很多。

下面分析代码。

17.1.1　依序下载的脚本

示例 17-2 不太有吸引力，不过实现并发下载的脚本时会重用其中的大部分代码和设置，因此值得分析一下。

　为了清楚起见，示例 17-2 没有处理异常。稍后会处理异常，这里我们想集中说明代码的基本结构，以便和并发下载的脚本进行对比。

示例 17-2　flags.py：依序下载的脚本；另外两个脚本会重用其中几个函数

import os
import time
import sys

import requests  ➊

POP20_CC = ('CN IN US ID BR PK NG BD RU JP '
      'MX PH VN ET EG DE IR TR CD FR').split()  ➋

BASE_URL = 'http://flupy.org/data/flags'  ➌

DEST_DIR = 'downloads/'  ➍


def save_flag(img, filename):  ➎
  path = os.path.join(DEST_DIR, filename)
  with open(path, 'wb') as fp:
    fp.write(img)


def get_flag(cc):  ➏
  url = '{}/{cc}/{cc}.gif'.format(BASE_URL, cc=cc.lower())
  resp = requests.get(url)
  return resp.content


def show(text):  ➐
  print(text, end=' ')
  sys.stdout.flush()


def download_many(cc_list):  ➑
  for cc in sorted(cc_list):  ➒
    image = get_flag(cc)
    show(cc)
    save_flag(image, cc.lower() + '.gif')

  return len(cc_list)


def main(download_many):  ➓
  t0 = time.time()
  count = download_many(POP20_CC)
  elapsed = time.time() - t0
  msg = '\n{} flags downloaded in {:.2f}s'
  print(msg.format(count, elapsed))


if __name__ == '__main__':
  main(download_many)  ⓫

❶ 导入 requests 库。这个库不在标准库中，因此依照惯例，在导入标准库中的模块（os、time 和 sys）之后导入，而且使用一个空行分隔开。3

3可以使用 pip install requests 命令安装 requests 库。——编者注

❷ 列出人口最多的 20 个国家的 ISO 3166 国家代码，按照人口数量降序排列。

❸ 获取国旗图像的网站。4

4国旗图像出自 CIA 世界概况，由美国政府发布，属公共领域。我把这些图像复制到了自己的网站，以此避免向 CIA.gov 发起 DoS 攻击的嫌疑。

❹ 保存图像的本地目录。

❺ 把 img（字节序列）保存到 DEST_DIR 目录中，命名为 filename。

❻ 指定国家代码，构建 URL，然后下载图像，返回响应中的二进制内容。

❼ 显示一个字符串，然后刷新 sys.stdout，这样能在一行消息中看到进度。在 Python 中得这么做，因为正常情况下，遇到换行才会刷新 stdout 缓冲。

❽ download_many 是与并发实现比较的关键函数。

❾ 按字母表顺序迭代国家代码列表，明确表明输出的顺序与输入一致。返回下载的国旗数量。

❿ main 函数记录并报告运行 download_many 函数之后的耗时。

⓫ main 函数必须调用执行下载的函数；我们把 download_many 函数当作参数传给 main 函数，这样 main 函数可以用作库函数，在后面的示例中接收 download_many 函数的其他实现。

　Kenneth Reitz 开发的 requests 库可通过 PyPI 安装，比 Python 3 标准库中的 urllib.request 模块更易于使用。其实，requests 库提供的 API 更符合 Python 的习惯用法，而且与 Python 2.6 及以上版本兼容。因为 Python 2 中删除了 urllib2，Python 3 又使用了其他名称，所以不管使用哪一版 Python，使用 requests 库都更方便。

flags.py 脚本中没有什么新知识，只是与其他脚本对比的基准，而且我把它作为一个库使用，避免实现其他脚本时重复编写代码。下面分析使用 concurrent.futures 模块重新实现的版本。

17.1.2　使用concurrent.futures模块下载

concurrent.futures 模块的主要特色是 ThreadPoolExecutor 和 ProcessPoolExecutor 类，这两个类实现的接口能分别在不同的线程或进程中执行可调用的对象。这两个类在内部维护着一个工作线程或进程池，以及要执行的任务队列。不过，这个接口抽象的层级很高，像下载国旗这种简单的案例，无需关心任何实现细节。

示例 17-3 展示如何使用 ThreadPoolExecutor.map 方法，以最简单的方式实现并发下载。

示例 17-3　flags_threadpool.py：使用 futures.ThreadPoolExecutor 类实现多线程下载的脚本

from concurrent import futures

from flags import save_flag, get_flag, show, main  ➊

MAX_WORKERS = 20  ➋


def download_one(cc):  ➌
  image = get_flag(cc)
  show(cc)
  save_flag(image, cc.lower() + '.gif')
  return cc


def download_many(cc_list):
  workers = min(MAX_WORKERS, len(cc_list))  ➍
  with futures.ThreadPoolExecutor(workers) as executor:  ➎
    res = executor.map(download_one, sorted(cc_list))  ➏

  return len(list(res))  ➐


if __name__ == '__main__':
  main(download_many)  ➑

❶ 重用 flags 模块（见示例 17-2）中的几个函数。

❷ 设定 ThreadPoolExecutor 类最多使用几个线程。

❸ 下载一个图像的函数；这是在各个线程中执行的函数。

❹ 设定工作的线程数量：使用允许的最大值（MAX_WORKERS）与要处理的数量之间较小的那个值，以免创建多余的线程。

❺ 使用工作的线程数实例化 ThreadPoolExecutor 类；executor.__exit__ 方法会调用 executor.shutdown(wait=True) 方法，它会在所有线程都执行完毕前阻塞线程。

❻ map 方法的作用与内置的 map 函数类似，不过 download_one 函数会在多个线程中并发调用；map 方法返回一个生成器，因此可以迭代，获取各个函数返回的值。

❼ 返回获取的结果数量；如果有线程抛出异常，异常会在这里抛出，这与隐式调用 next() 函数从迭代器中获取相应的返回值一样。

❽ 调用 flags 模块中的 main 函数，传入 download_many 函数的增强版。

注意，示例 17-3 中的 download_one 函数其实是示例 17-2 中 download_many 函数的 for 循环体。编写并发代码时经常这样重构：把依序执行的 for 循环体改成函数，以便并发调用。

我们用的库叫 concurrent.futures，可是在示例 17-3 中没有见到期物，因此你可能想知道期物在哪里。下一节会解答这个问题。

17.1.3　期物在哪里

期物是 concurrent.futures 模块和 asyncio 包的重要组件，可是，作为这两个库的用户，我们有时却见不到期物。示例 17-3 在背后用到了期物，但是我编写的代码没有直接使用。这一节概述期物，还会举一个例子，展示用法。

从 Python 3.4 起，标准库中有两个名为 Future 的类：concurrent.futures.Future 和 asyncio.Future。这两个类的作用相同：两个 Future 类的实例都表示可能已经完成或者尚未完成的延迟计算。这与 Twisted 引擎中的 Deferred 类、Tornado 框架中的 Future 类，以及多个 JavaScript 库中的 Promise 对象类似。

期物封装待完成的操作，可以放入队列，完成的状态可以查询，得到结果（或抛出异常）后可以获取结果（或异常）。

我们要记住一件事：通常情况下自己不应该创建期物，而只能由并发框架（concurrent.futures 或 asyncio）实例化。原因很简单：期物表示终将发生的事情，而确定某件事会发生的唯一方式是执行的时间已经排定。因此，只有排定把某件事交给 concurrent.futures.Executor 子类处理时，才会创建 concurrent.futures.Future 实例。例如，Executor.submit() 方法的参数是一个可调用的对象，调用这个方法后会为传入的可调用对象排期，并返回一个期物。

客户端代码不应该改变期物的状态，并发框架在期物表示的延迟计算结束后会改变期物的状态，而我们无法控制计算何时结束。

这两种期物都有 .done() 方法，这个方法不阻塞，返回值是布尔值，指明期物链接的可调用对象是否已经执行。客户端代码通常不会询问期物是否运行结束，而是会等待通知。因此，两个 Future 类都有 .add_done_callback() 方法：这个方法只有一个参数，类型是可调用的对象，期物运行结束后会调用指定的可调用对象。

此外，还有 .result() 方法。在期物运行结束后调用的话，这个方法在两个 Future 类中的作用相同：返回可调用对象的结果，或者重新抛出执行可调用的对象时抛出的异常。可是，如果期物没有运行结束，result 方法在两个 Future 类中的行为相差很大。对 concurrency.futures.Future 实例来说，调用 f.result() 方法会阻塞调用方所在的线程，直到有结果可返回。此时，result 方法可以接收可选的 timeout 参数，如果在指定的时间内期物没有运行完毕，会抛出 TimeoutError 异常。读到 18.1.1 节你会发现，asyncio.Future.result 方法不支持设定超时时间，在那个库中获取期物的结果最好使用 yield from 结构。不过，对 concurrency.futures.Future 实例不能这么做。

这两个库中有几个函数会返回期物，其他函数则使用期物，以用户易于理解的方式实现自身。使用 17-3 中的 Executor.map 方法属于后者：返回值是一个迭代器，迭代器的 __next__ 方法调用各个期物的 result 方法，因此我们得到的是各个期物的结果，而非期物本身。

为了从实用的角度理解期物，我们可以使用 concurrent.futures.as_completed 函数重写示例 17-3。这个函数的参数是一个期物列表，返回值是一个迭代器，在期物运行结束后产出期物。

为了使用 futures.as_completed 函数，只需修改 download_many 函数，把较抽象的 executor.map 调用换成两个 for 循环：一个用于创建并排定期物，另一个用于获取期物的结果。同时，我们会添加几个 print 调用，显示运行结束前后的期物。修改后的 download_many 函数如示例 17-4，代码行数由 5 变成 17，不过现在我们能一窥神秘的期物了。其他函数不变，与示例 17-3 中的一样。

示例 17-4　flags_threadpool_ac.py：把download_many 函数中的 executor.map 方法换成 executor.submit 方法和 futures.as_completed 函数

def download_many(cc_list):
  cc_list = cc_list[:5]  ➊
  with futures.ThreadPoolExecutor(max_workers=3) as executor:  ➋
    to_do = []
    for cc in sorted(cc_list):  ➌
      future = executor.submit(download_one, cc)  ➍
      to_do.append(future)  ➎
      msg = 'Scheduled for {}: {}'
      print(msg.format(cc, future))  ➏

    results = []
    for future in futures.as_completed(to_do):  ➐
      res = future.result()  ➑
      msg = '{} result: {!r}'
      print(msg.format(future, res))  ➒
      results.append(res)

  return len(results)

❶ 这次演示只使用人口最多的 5 个国家。

❷ 把 max_workers 硬编码为 3，以便在输出中观察待完成的期物。

❸ 按照字母表顺序迭代国家代码，明确表明输出的顺序与输入一致。

❹ executor.submit 方法排定可调用对象的执行时间，然后返回一个期物，表示这个待执行的操作。

❺ 存储各个期物，后面传给 as_completed 函数。

❻ 显示一个消息，包含国家代码和对应的期物。

❼ as_completed 函数在期物运行结束后产出期物。

❽ 获取该期物的结果。

❾ 显示期物及其结果。

注意，在这个示例中调用 future.result() 方法绝不会阻塞，因为 future 由 as_completed 函数产出。运行示例 17-4 得到的输出如示例 17-5 所示。

示例 17-5　flags_threadpool_ac.py 脚本的输出

$ python3 flags_threadpool_ac.py
Scheduled for BR: <Future at 0x100791518 state=running> ➊
Scheduled for CN: <Future at 0x100791710 state=running>
Scheduled for ID: <Future at 0x100791a90 state=running>
Scheduled for IN: <Future at 0x101807080 state=pending> ➋
Scheduled for US: <Future at 0x101807128 state=pending>
CN <Future at 0x100791710 state=finished returned str> result: 'CN' ➌
BR ID <Future at 0x100791518 state=finished returned str> result: 'BR' ➍
<Future at 0x100791a90 state=finished returned str> result: 'ID'
IN <Future at 0x101807080 state=finished returned str> result: 'IN'
US <Future at 0x101807128 state=finished returned str> result: 'US'

5 flags downloaded in 0.70s

❶ 排定的期物按字母表排序；期物的 repr() 方法会显示期物的状态：前三个期物的状态是 running，因为有三个工作的线程。

❷ 后两个期物的状态是 pending，等待有线程可用。

❸ 这一行里的第一个 CN 是运行在一个工作线程中的 download_one 函数输出的，随后的内容是 download_many 函数输出的。

❹ 这里有两个线程输出国家代码，然后主线程中的 download_many 函数输出第一个线程的结果。

　多次运行 flags_threadpool_ac.py 脚本，看到的结果有所不同。如果把 max_workers 参数的值增大到 5，结果的顺序变化更多。把 max_workers 参数的值设为 1，代码依序运行，结果的顺序始终与调用 submit 方法的顺序一致。

我们分析了两个版本的使用 concurrent.futures 库实现的下载脚本：使用 ThreadPoolExecutor.map 方法的示例 17-3 和使用 futures.as_completed 函数的示例 17-4。如果你对 flags_asyncio.py 脚本的代码好奇，可以看一眼第 18 章中的示例 18-5。

严格来说，我们目前测试的并发脚本都不能并行下载。使用 concurrent.futures 库实现的那两个示例受 GIL（Global Interpreter Lock，全局解释器锁）的限制，而 flags_asyncio.py 脚本在单个线程中运行。

读到这里，你可能会对前面做的非正规基准测试有下述疑问。

既然 Python 线程受 GIL 的限制，任何时候都只允许运行一个线程，那么 flags_threadpool.py 脚本的下载速度怎么会比 flags.py 脚本快 5 倍？

flags_asyncio.py 脚本和 flags.py 脚本都在单个线程中运行，前者怎么会比后者快 5 倍？

第二个问题在 18.3 节解答。

GIL 几乎对 I/O 密集型处理无害，原因参见下一节。