Fluent Python - Part17使用期物处理并发_综合

本章主要讨论两点

Python3.2 引入的 concurrent.futures 模块
期物(future)的概念。期物指一种对象，表示异步执行的操作。这个概念的作用很大，是 concurrent.futures 模块和 asyncio 包的基础。

示例：网络下载的三种风格

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

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

依序下载的脚本

import os
import time
import sysimport requestsPOP20_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.contentdef 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() - t0msg = '\n{} flags downloaded in {:.2f}s'print(msg.format(count, elapsed))if __name__ == '__main__':main(download_many)""" output: BD BR CD CN DE EG ET FR ID IN IR JP MX NG PH PK RU TR US VN 20 flags downloaded in 24.41s """

使用 `concurrent.futures` 模块下载

concurrent.futures 模块的主要特色是 ThreadPoolExecutor 和 ProcessPoolExecutor 类，这两个类实现的接口能分别在不同的线程或进程中执行可调用的对象。这两个类在内部维护着一个工作线程或进程池，以及要执行的任务队列。

import os
import time
import sysimport requests
from concurrent import futuresPOP20_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.contentdef show(text):print(text, end=' ')sys.stdout.flush()def download_one(cc):image = get_flag(cc)show(cc)save_flag(image, cc.lower() + '.gif')return ccMAX_WORKERS = 20
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))def main(download_many):t0 = time.time()count = download_many(POP20_CC)elapsed = time.time() - t0msg = '\n{} flags downloaded in {:.2f}s'print(msg.format(count, elapsed))if __name__ == '__main__':main(download_many)""" output: BD FR EG NG IR DE CD US IN PK RU ID BR TR VN PH MX ET CN JP 20 flags downloaded in 0.76s """

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

期物在哪里

期物是 concurrent.futures 模块和 asyncio 包的重要组件。从 Python3.4 起，标准库中有两个名为 Future 的类：concurrent.futures.Future 和 asyncio.Future。这两个类的作用相同：两个 Future 类的实例都表示可能已经完成或者尚未完成的延迟计算。

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

通常情况下不应该自己创建期物，而只能由并发框架(concurrent.futures 或 asyncio) 实例化。客户端代码也不应该改变期物的状态。

这两个期物都有 .done() 方法，这个方法不阻塞，返回值是布尔值，指明期物链接的可调用对象是否已经执行。两个 Future 类都用 .add_done_callback() 方法：这个方法只有一个参数，类型是可调用的对象，期物运行结束后会调用指定的可调用对象。

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

为了从实用的角度理解期物，我们可以使用 concurrent.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)""" output: Scheduled for BR: <Future at 0x29793bc2cc8 state=running> Scheduled for CN: <Future at 0x29793bd2388 state=running> Scheduled for ID: <Future at 0x29793bd2d88 state=running> Scheduled for IN: <Future at 0x29793bda848 state=pending> Scheduled for US: <Future at 0x29793bda908 state=pending> BR <Future at 0x29793bc2cc8 state=finished returned str> result: 'BR' CN <Future at 0x29793bd2388 state=finished returned str> result: 'CN' ID <Future at 0x29793bd2d88 state=finished returned str> result: 'ID' US <Future at 0x29793bda908 state=finished returned str> result: 'US' IN <Future at 0x29793bda848 state=finished returned str> result: 'IN'5 flags downloaded in 8.21sProcess finished with exit code 0"""

阻塞型I/O和GIL

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

标准库中所有执行阻塞型 I/O 操作的函数，在等待操作系统返回结果时都会释放 GIL。这意味着在 Python 语言这个层次上可以使用多线程，而 I/O 密集型 Python 程序能从中受益。

下面简单说明如何在 CPU 密集型作业中使用 concurrent.futures 模块轻松绕开 GIL。

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

ProcessorPoolExecutor 和 ThreadPoolExecutor 类都实现了通用的 Executor 接口.因此使用 concurrent.futures 模块能特别轻松地把基于线程的方案转成基于进程的方案。

def download_many(cc_list):    with futures.ProcessPoolExecutor() as executor:

实验 `Executor.map` 方法

from time import sleep, strftime
from concurrent import futuresdef display(*args):print(strftime('[%H:%M:%S]'), end=' ')print(*args)def loiter(n):msg = '{}loiter({}): doing nothing for {}s...'display(msg.format('\t'*n, n, n))sleep(n)msg = '{}loiter({}): done'display(msg.format('\t'*n, n))return n * 10def main():display('Script starting')executor = futures.ThreadPoolExecutor(max_workers=3)results = executor.map(loiter, range(5))display('results:', results)display('Waiting for individual results:')for i, result in enumerate(results):display('result {}: {}'.format(i, result))main()""" output: [21:57:27] Script starting [21:57:27] loiter(0): doing nothing for 0s... [21:57:27] loiter(0): done [21:57:27] loiter(1): doing nothing for 1s... [21:57:27] loiter(2): doing nothing for 2s... [21:57:27] results: <generator object Executor.map.<locals>.result_iterator at 0x000001A9D7B93AC8> [21:57:27] Waiting for individual results: [21:57:27] result 0: 0 [21:57:27] loiter(3): doing nothing for 3s... [21:57:28] loiter(1): done [21:57:28] loiter(4): doing nothing for 4s... [21:57:28] result 1: 10 [21:57:29] loiter(2): done [21:57:29] result 2: 20 [21:57:30] loiter(3): done [21:57:30] result 3: 30 [21:57:32] loiter(4): done [21:57:32] result 4: 40Process finished with exit code 0"""

下一章我们将介绍“使用 asyncio 包处理并发”

Fluent Python - Part17使用期物处理并发