使用asyncio协同程序的Web爬虫

A. Jesse Jiryu Davis和Guido van Rossum

A. Jesse Jiryu Davis是纽约MongoDB的一名工程师。他写了Motor，这是异步的MongoDB Python驱动程序，他是MongoDB C Driver的首席开发人员，也是PyMongo团队的成员。他贡献asyncio和龙卷风。他在http://emptysqua.re写道。

Guido van Rossum是Python的创建者，Python是网络内外的主要编程语言之一。Python社区称他为BDFL（仁慈的生命独裁者），直接来自Monty Python的小品。

介绍

经典的计算机科学强调高效的算法，尽可能快地完成计算。但是许多联网程序花费的时间不是计算，而是保持打开许多缓慢连接或偶尔发生事件的连接。这些计划提出了一个非常不同的挑战：高效地等待大量网络事件。当前解决这个问题的方法是异步I / O或“异步”。

本章介绍一个简单的网络爬虫。抓取工具是一个典型的异步应用程序，因为它会等待很多响应，但计算量很小。它可以一次获取的页面越多，它就越快完成。如果它为每个正在进行的请求提供一个线程，那么随着并发请求的数量的增加，它在耗尽socket之前将耗尽内存或其他线程相关资源。它通过使用异步I / O避免了线程的需要。

我们分三个阶段来举例。首先，我们展示一个异步事件循环，并绘制一个使用回调事件循环的爬虫程序：它非常高效，但将其扩展到更复杂的问题将导致无法管理的意大利面代码。其次，因此，我们展示Python协程既高效又可扩展。我们使用生成器函数在Python中实现简单的协程。在第三阶段，我们使用Python标准“asyncio”库的全功能协程，并使用异步队列协调它们。

任务

网络爬虫可以查找并下载网站上的所有网页，也许可以将其归档或编入索引。从根URL开始，它会获取每个页面，解析它以查找未见页面的链接，并将这些页面添加到队列中。它在获取没有看不见的链接并且队列为空的页面时停止。

我们可以通过同时下载多个页面来加速这一过程。当抓取工具发现新的链接时，它会在单独的socket上为新页面启动同时抓取操作。它在到达时解析响应，并向队列添加新的链接。可能会出现一些收益递减的情况，即过多的并发性会降低性能，因此我们限制并发请求的数量，并将剩余的链接留在队列中，直到某些正在进行的请求完成为止。

传统方法

我们如何使爬行器并发？传统上我们会创建一个线程池。每个线程将负责通过socket一次下载一个页面。例如，要从xkcd.com以下网址下载页面：

def fetch(url):sock = socket.socket()sock.connect(('xkcd.com', 80))request = 'GET {} HTTP/1.0\r\nHost: xkcd.com\r\n\r\n'.format(url)sock.send(request.encode('ascii'))response = b''chunk = sock.recv(4096)while chunk:response += chunkchunk = sock.recv(4096)# Page is now downloaded.links = parse_links(response)q.add(links)

默认情况下，socket操作是阻塞的：当线程调用像connector recv这样的方法时，它会暂停，直到操作完成。因此，一次下载许多页面，我们需要很多线程。一个复杂的应用程序通过保持线程池中的空闲线程来分摊线程创建的成本，然后将它们检出并重用以用于后续任务; 它与连接池中的socket一样。

然而，线程很昂贵，并且操作系统对进程，用户或机器可能具有的线程数量强制执行各种硬性限制。在Jesse的系统中，Python线程花费大约50k的内存，并且启动成千上万的线程会导致失败。如果我们在并发socket上扩展到数万个并发操作，那么在socket用完之前我们会耗尽线程。线程的每线程开销或系统限制是瓶颈。

在他有影响力的文章“The C10K problem” ，Dan Kegel概述了多线程对I / O并发性的局限性。他开始，

现在是Web服务器同时处理一万个客户端的时候了，你不觉得吗？毕竟，网络现在是一个很大的地方。

凯格尔在1999年创造了“C10K”这个术语。现在，有一万个连接听起来很美观，但问题只是在大小上改变，而不是实物。那时候，为C10K使用每个连接的线程是不切实际的。现在帽子的数量级要高一些。事实上，我们的玩具网络爬虫可以在线程中正常工作。然而，对于具有数十万个连接的超大规模应用程序来说，这个上限依然存在：超过这个限制，大多数系统仍然可以创建socket，但已经没有线程了。我们怎样才能克服这一点？

异步

异步I / O框架使用非阻塞socket对单个线程执行并发操作。在我们的异步爬虫中，我们在开始连接到服务器之前设置socket非阻塞：

sock = socket.socket()
sock.setblocking(False)
try:sock.connect(('xkcd.com', 80))
except BlockingIOError:pass

令人恼火的是connect，即使正常工作，非阻塞socket也会引发异常。此异常复制下面的C函数，它设置的刺激性行为errno来EINPROGRESS告诉你它已经开始。

现在我们的爬虫需要一种方法来了解何时建立连接，因此它可以发送HTTP请求。我们可以简单地继续努力：

request = 'GET {} HTTP/1.0\r\nHost: xkcd.com\r\n\r\n'.format(url)
encoded = request.encode('ascii')while True:try:sock.send(encoded)break  # Done.except OSError as e:passprint('sent')

这种方法不仅浪费电能，但它不能有效地等待关于事件的多个插座。在古代，BSD Unix的解决这一问题的是select，一个C函数等待非阻塞插座或其中的一小阵列上发生的事件。如今与连接的数量巨大的互联网应用的需求已经导致了像更换poll，然后kqueue在BSD和epollLinux的上。这些API类似select，但具有非常大量的连接表现良好。

Python的3.4的DefaultSelector使用最好的select系统上可用般的功能。要注册有关网络我的通知/ O，我们创建了一个非阻塞socket，并默认选择注册它：

from selectors import DefaultSelector, EVENT_WRITEselector = DefaultSelector()sock = socket.socket()
sock.setblocking(False)
try:sock.connect(('xkcd.com', 80))
except BlockingIOError:passdef connected():selector.unregister(sock.fileno())print('connected!')selector.register(sock.fileno(), EVENT_WRITE, connected)

我们不顾虚假错误并调用selector.register，传递socket的文件描述符和表示，我们正在等待什么事件为一个常数。将通知的建立连接时，我们通过EVENT_WRITE：那就是，我们要知道什么时候该插座是“可写”。我们也通过一个Python的函数，connected以当事件发生时运行。这种功能就是所谓的回调。

我们处理I / O通知作为选择接收他们，在一个循环：

def loop():while True:events = selector.select()for event_key, event_mask in events:callback = event_key.datacallback()

该connected回调存储event_key.data，这是我们检索和执行一次非阻塞socket连接。

不像在我们上面的快速自转循环，调用select这里停了下来，等待着下一个I / O事件。然后循环运行正在等待这些事件的回调。还没有完成的操作保持挂起，直到一些未来事件循环的滴答声。

我们用什么证明了吗？我们展示了如何在动作准备开始操作和执行的回调。异步架构建立在两个特征我们已经表明，非阻塞socket和事件循环，在单个线程中运行并发操作。

我们已经取得了“并发”在这里，但不是传统上称为“并行”。也就是说，我们建立了不重叠I / O一个微小的系统。它能够开始新的行动，而别人都在飞行。它实际上并没有利用多内核并行执行计算。但随后，该系统是专为I / O密集型的问题，而不是CPU绑定的。

因此，我们的事件循环是并行I / O效率，因为它不投入线程资源给每个连接。但是，在我们开始之前，纠正一个误解通用异步这是很重要更快比多线程。通常它不是，事实上，在Python的中，一个事件循环，像我们这样的中等比服务于少数非常活跃的连接的多线程慢。在没有全局解释锁运行时，线程将执行即使是在这样的工作量更好。异步I / O是正确的，是与偶发事件许多慢或困连接的应用程序。

编程使用回调

随着runty异步架构到目前为止，我们已经建立了，我们怎么可以建立一个网络爬虫？即使是一个简单的URL取功能是很痛苦的决定。

我们开始我们还没有获取网址，并且我们已经看到了网址的全球集：

urls_todo = set(['/'])
seen_urls = set(['/'])

该seen_urls集包括urls_todo加完成的URL。两套初始化与根URL“/”。

抓取的网页，需要一系列的回调。该connected回调火灾时，一个socket连接，并发送一GET请求到服务器。但是，那么它必须等待响应，所以它注册另一个回调。如果，当回调火灾，它无法读取完整的回应是，它再次注册，等等。

我们让这些收集到回调一个Fetcher对象。它需要一个网址，一个插座对象，并且累积响应字节的地方：

class Fetcher:def __init__(self, url):self.response = b''  # Empty array of bytes.self.url = urlself.sock = None

我们首先通过调用Fetcher.fetch：

    # Method on Fetcher class.def fetch(self):self.sock = socket.socket()self.sock.setblocking(False)try:self.sock.connect(('xkcd.com', 80))except BlockingIOError:pass# Register next callback.selector.register(self.sock.fileno(),EVENT_WRITE,self.connected)

该fetch方法开始连接的插座。但是请注意，该方法返回建立连接之前，它必须将控制返回给事件循环等待连接。要理解为什么，想象我们整个应用程序的结构是这样：

# Begin fetching http://xkcd.com/353/
fetcher = Fetcher('/353/')
fetcher.fetch()while True:events = selector.select()for event_key, event_mask in events:callback = event_key.datacallback(event_key, event_mask)

所有事件通知在事件处理循环当它调用select。因此，fetch必须用手控制事件循环，这样，当socket连接程序知道。之后才进行循环运行的connected回调，的英文这在年底注册fetch的上方。

下面是执行connected：

    # Method on Fetcher class.def connected(self, key, mask):print('connected!')selector.unregister(key.fd)request = 'GET {} HTTP/1.0\r\nHost: xkcd.com\r\n\r\n'.format(self.url)self.sock.send(request.encode('ascii'))# Register the next callback.selector.register(key.fd,EVENT_READ,self.read_response)

的方法发送的GET请求。实际的应用程序会检查报道查看值send的情况下，整个消息不能发送一次。但是，我们的要求是小，我们的应用程序不复杂，它轻率地调用send，然后等待响应。当然，它必须注册另一个回调放弃控制事件循环。第二天和最后回调read_response，处理服务器的答复：

    # Method on Fetcher class.def read_response(self, key, mask):global stoppedchunk = self.sock.recv(4096)  # 4k chunk size.if chunk:self.response += chunkelse:selector.unregister(key.fd)  # Done reading.links = self.parse_links()# Python set-logic:for link in links.difference(seen_urls):urls_todo.add(link)Fetcher(link).fetch()  # <- New Fetcher.seen_urls.update(links)urls_todo.remove(self.url)if not urls_todo:stopped = True

回调各个选择看到该插座是“可读”的时间，这可能意味着两件事情执行：socket有数据或者它是封闭的。

回调要求从插座数据的多达四个千字节。如果少准备，chunk包含任何数据是可用的。如果有更多的，chunk为四字节长和插座保持可读，所以事件循环再次运行此回调的下一个节拍。当反应完成后，服务器已关闭socket，并chunk为空。

的parse_links方法，未示出，返回一组网址。我们开始新的提取器为每一个新的URL，没有并发上限。注意与回调异步编程的一个很好的功能：我们需要改变周围没有互斥共享数据，比如当我们添加链接seen_urls。没有抢占式多任务，所以我们不能在我们的代码中的任意点中断。

一个添加全局stopped变量，并用它来控制循环：

stopped = Falsedef loop():while not stopped:events = selector.select()for event_key, event_mask in events:callback = event_key.datacallback()

一旦所有的网页下载取出器将停止全局事件循环和退出程序。

这个例子使普通异步的问题：意大利面条代码。我们需要一些方法来表达一系列的计算和I / O操作，并多个安排这样的一系列操作的同时运行但是，如果没有线程，一系列的操作不能被收集到一个单一的功能：当函数开头的I / O操作，它明确地保存任何状态将在未来的需要，然后返回。你是负责思考和写这个状态保存代码。

让我们来解释一下我们的意思了。考虑如何简单地拿来与传统的阻塞socket线程的网址：

# Blocking version.
def fetch(url):sock = socket.socket()sock.connect(('xkcd.com', 80))request = 'GET {} HTTP/1.0\r\nHost: xkcd.com\r\n\r\n'.format(url)sock.send(request.encode('ascii'))response = b''chunk = sock.recv(4096)while chunk:response += chunkchunk = sock.recv(4096)# Page is now downloaded.links = parse_links(response)q.add(links)

，这种功能一个插座操作和未来之间还记得是什么状态？它有插座，URL和积累response。上一个线程中运行的功能使用的编程语言的基本功能，存储在本地变量这种临时状态，它的堆栈。该功能也有一个“延续”，也就是说，它计划后，我来执行代码/ O完成。运行时通过存储线程的指令指针记得延续。你不必去想以后的I / O恢复这些局部变量和延续。它是建立在对语言。

但是随着基于回调的异步框架，这些语言的特点是没有任何帮助。在等待I / O，一个函数必须明确地保存它的状态，因为在函数返回之前的I / O完成失去它的包栈帧。代替局部变量，基于我们的回调例子卖场sock状语从句：response它们的属性self，在提取程序实例。代替指令指针，通过它注册回调存储其延续connected状语从句：read_response。作为应用程序的功能增长，将在国家我们遇到回调手动保存的复杂性。这样繁重的簿记使编码器容易出现偏头痛。

更糟的是，如果回调链中抛出一个异常，才调度下一个回调会发生什么？说我们做的差的工作parse_links方法，它抛出一个异常解析一些HTML：

Traceback (most recent call last):File "loop-with-callbacks.py", line 111, in <module>loop()File "loop-with-callbacks.py", line 106, in loopcallback(event_key, event_mask)File "loop-with-callbacks.py", line 51, in read_responselinks = self.parse_links()File "loop-with-callbacks.py", line 67, in parse_linksraise Exception('parse error')
Exception: parse error

堆栈跟踪仅显示事件循环正在运行的回调。我们不记得是什么导致了错误。链被打破两端：我们忘记了我们要去哪里，并那里从我们来到上下文的这种损失被称为“堆翻录”，而且在许多情况下，它混淆了调查。堆栈翻录也阻止了我们安装一个异常处理程序回调链，顺便‘尝试/除’块包装了一个函数调用及其后代的树。

因此，即使从约多线程和异步的相对效率的长期争论之余，有关于这是比较容易出错的这个其他的争论：如果你犯了一个错误使其同步线程很容易受到数据的比赛，但回调固执调试由于堆栈翻录。

协同程序

我们吸引你的承诺。它是可以编写结合的回调与多线程编程的经典外型美观效率异步代码。这种组合与一个名为“协程”模式实现的。使用Python 3.4的标准ASYNCIO库，以及一个名为“aiohttp”包，一个协程提取网址是非常直接的：

    @asyncio.coroutinedef fetch(self, url):response = yield from self.session.get(url)body = yield from response.read()

这也是可扩展的。相比于每个线程的内存和线程操作系统的硬性限制的50K，一个Python协程需要勉强3K的内存杰西的系统上。Python可以轻松地开始数十万协同程序。

协程的概念，可以追溯到计算机科学的长老天，很简单：它是可以暂停和恢复的子程序而线程抢先操作系统多任务，多任务协同程序合作：他们选择什么时候暂停，并协同程序运行下。

有协同程序的许多实现; 即使在Python的有以下几种。在在Python的3.4标准的“ASYNCIO”库中的协同程序是在发电机，一个未来类，并声明“从收益率”建成。在3.5的Python开始，协程是语言本身的一项内置功能; 然而，了解协同程序，因为他们在Python 3.4开始实施，使用预先存在的语言工具，是解决的Python 3.5的原生协程的基础。

为了解释的Python 3.4的基于生成器的协同程序，我们将参与发电机的论述以及它们如何用作ASYNCIO协同程序，相信你会喜欢阅读它，就像我们喜欢写它。一旦我们解释的发电的协同程序，我们将在我们的异步网络爬虫使用它们。

蟒蛇和发电机工作

在你掌握Python的生成器，你要明白的Python函数如何正常工作。通常情况下，当一个Python的函数调用子程序，子程序保留控制权，直到它返回，或抛出异常。然后控制返回给调用者：

>>> def foo():
...     bar()
...
>>> def bar():
...     pass

标准的Python解释被写入C. ç函数执行一个Python的函数被调用，mellifluously， PyEval_EvalFrameEx。这需要一个Python的堆栈帧对象和在该帧的上下文中评估的Python的字节码。这里是字节代码foo：

>>> import dis
>>> dis.dis(foo)2           0 LOAD_GLOBAL              0 (bar)3 CALL_FUNCTION            0 (0 positional, 0 keyword pair)6 POP_TOP7 LOAD_CONST               0 (None)10 RETURN_VALUE

该foo函数加载bar到其堆栈并调用它，然后从栈中弹出它的返回值，负荷None压入堆栈，并返回None。

当PyEval_EvalFrameEx遇到CALL_FUNCTION的字节码时，它创建一个新的Python的堆栈帧和递归：即，调用它PyEval_EvalFrameEx与新的帧，其被用来执行递归bar。

关键是要明白的Python栈帧是在堆内存分配！Python解释是一个普通的C程序，所以它的栈帧是正常的堆栈帧。但是Python的堆栈帧其所有的操作的因素，堆。在其他的意外，这意味着一个Python的堆栈帧可以活得比其函数调用。要交互式地看到这一点，从内部保存当前帧bar：

>>> import inspect
>>> frame = None
>>> def foo():
...     bar()
...
>>> def bar():
...     global frame
...     frame = inspect.currentframe()
...
>>> foo()
>>> # The frame was executing the code for 'bar'.
>>> frame.f_code.co_name
'bar'
>>> # Its back pointer refers to the frame for 'foo'.
>>> caller_frame = frame.f_back
>>> caller_frame.f_code.co_name
'foo'

图5.1 - 函数调用

阶段现在被设置为Python的发电机，其中使用相同的构建块码的对象和堆栈帧到奇妙的效果。

这是一个发生器功能：

>>> def gen_fn():
...     result = yield 1
...     print('result of yield: {}'.format(result))
...     result2 = yield 2
...     print('result of 2nd yield: {}'.format(result2))
...     return 'done'
...     

当Python的compile- gen_fn分类中翻译字节码，看到它的yield语句，并知道gen_fn的英文发电机的功能，而不是常规的。它设置一个标志要记住这样一个事实：

>>> # The generator flag is bit position 5.
>>> generator_bit = 1 << 5
>>> bool(gen_fn.__code__.co_flags & generator_bit)
True

当你调用生成函数，巨蟒看到发电机标志，它不实际运行的功能。相反，它创建了一个发电机：

>>> gen = gen_fn()
>>> type(gen)
<class 'generator'>

一个Python的发电机封装了一个堆栈帧加上一些代码的引用，所述主体的gen_fn：

>>> gen.gi_code.co_name
'gen_fn'

的所有调用发电机gen_fn指向相同的代码。但是，每个人都有自己的堆栈帧。此堆栈帧不是在任何实际堆，它坐落在堆内存等待被使用：

图5.2 - 发电机

该框架有一个“最后一条指令”指针，它最近执行的指令。在开始的时候，最后的指令指针为-1，这意味着发电机还没有开始：

>>> gen.gi_frame.f_lasti
-1

当我们调用send，发电机达到其第一yield，并暂停。的报道查看值send的英文1，这因为的英文gen传递给yield表达式：

>>> gen.send(None)
1

发电机的指令指针现在是从一开始3个字节码，通过56个字节编译的Python中的一部分方式：

>>> gen.gi_frame.f_lasti
3
>>> len(gen.gi_code.co_code)
56

发电机可以在任何时候重新开始，从任何功能，因为它的栈帧是不实际的堆栈上。的英文它在堆上它在调用层次中的位置的英文不固定的，并且不需要服从执行的先入后出顺序，经常函数执行。据解放，自由漂浮，像云。

我们可以送价值“你好”到发电机，成为它的查询查询结果yield表达，以及发电机继续，直到产生2：

>>> gen.send('hello')
result of yield: hello
2

它的堆栈帧现在包含局部变量result：

>>> gen.gi_frame.f_locals
{'result': 'hello'}

创建³³从其他发电机gen_fn将有自己的堆栈帧状语从句：局部变量。

我们当调用send再次，发电机从它的第二延续yield，并提高通过特殊的完成StopIteration情况例外：

>>> gen.send('goodbye')
result of 2nd yield: goodbye
Traceback (most recent call last):File "<input>", line 1, in <module>
StopIteration: done

唯一的例外有一个值，它是发电机的返回值：字符串"done"。

构建协同程序搭配发电机

因此，发电机可以暂停，也可以用数值来恢复，而且它有一个返回值。听起来像是不错的原始赖以建立的异步编程模型，没有意大利面条回调！我们要建立一个“协同程序”：即协同计划与程序中的其它程序的程序。我们的协同程序将是那些在Python中的标准“ASYNCIO”库的简化版本。作为ASYNCIO，我们将使用发电机，期货，并声明“从收益率”。

首先，我们需要一种方式来表示一个协程正在等待一些未来的结果。精简版：

class Future:def __init__(self):self.result = Noneself._callbacks = []def add_done_callback(self, fn):self._callbacks.append(fn)def set_result(self, result):self.result = resultfor fn in self._callbacks:fn(self)

未来的最初“待定”。它是由一个叫“解决” set_result。

。我们让我们调整的抓取工具使用期货协同状语从句：程序我们写了fetch一个回调：

class Fetcher:def fetch(self):self.sock = socket.socket()self.sock.setblocking(False)try:self.sock.connect(('xkcd.com', 80))except BlockingIOError:passselector.register(self.sock.fileno(),EVENT_WRITE,self.connected)def connected(self, key, mask):print('connected!')# And so on....

该fetch方法开始连接的插座，然后注册回调，connected当插座准备好被执行。现在，我们可以在这两个步骤合并成一个协同程序：

    def fetch(self):sock = socket.socket()sock.setblocking(False)try:sock.connect(('xkcd.com', 80))except BlockingIOError:passf = Future()def on_connected():f.set_result(None)selector.register(sock.fileno(),EVENT_WRITE,on_connected)yield fselector.unregister(sock.fileno())print('connected!')

现在fetch的英文发电机的功能，而不是常规的一个，它因为所有游戏一个yield声明。我们创建了一个悬而未决的未来，然后产生它暂停fetch，直至插槽就绪。内在功能on_connected解决了未来。

但是，当未来的解决，有什么恢复的发电机？我们需要一个协程驱动程序。让我们把它叫做“任务”：

class Task:def __init__(self, coro):self.coro = corof = Future()f.set_result(None)self.step(f)def step(self, future):try:next_future = self.coro.send(future.result)except StopIteration:returnnext_future.add_done_callback(self.step)# Begin fetching http://xkcd.com/353/
fetcher = Fetcher('/353/')
Task(fetcher.fetch())loop()

开始任务fetch通过发送发电机None进去。然后fetch运行，直到它产生以后，它的任务是捕获next_future。当socket连接，事件循环运行的回调on_connected，它解决了未来，它调用step，从而恢复fetch。

保协程用 yield from

一旦socket连接，我们发送HTTP GET请求，并读取服务器响应。这些步骤不再需要被回调中四散。我们收集他们在同一个发生器功能：

    def fetch(self):# ... connection logic from above, then:sock.send(request.encode('ascii'))while True:f = Future()def on_readable():f.set_result(sock.recv(4096))selector.register(sock.fileno(),EVENT_READ,on_readable)chunk = yield fselector.unregister(sock.fileno())if chunk:self.response += chunkelse:# Done reading.break

此代码，它从一个socket读取整条消息，似乎通常是有用的。我们怎样才能从要它它fetch变成一个子程序？现在，Python 3中的着名yield from上台。它可以让一名发电机代表到另一个。

怎么看，让我们回到我们的简单生成器的例子：

>>> def gen_fn():
...     result = yield 1
...     print('result of yield: {}'.format(result))
...     result2 = yield 2
...     print('result of 2nd yield: {}'.format(result2))
...     return 'done'
...     

若要从另一个生成调用这个发电机，委托给它yield from：

>>> # Generator function:
>>> def caller_fn():
...     gen = gen_fn()
...     rv = yield from gen
...     print('return value of yield-from: {}'
...           .format(rv))
...
>>> # Make a generator from the
>>> # generator function.
>>> caller = caller_fn()

该caller发电机的作用英文就好像它的英文gen，它是委托给发电机：

>>> caller.send(None)
1
>>> caller.gi_frame.f_lasti
15
>>> caller.send('hello')
result of yield: hello
2
>>> caller.gi_frame.f_lasti  # Hasn't advanced.
15
>>> caller.send('goodbye')
result of 2nd yield: goodbye
return value of yield-from: done
Traceback (most recent call last):File "<input>", line 1, in <module>
StopIteration

虽然caller收益率从gen，caller不前进。请注意，它的指令指针保持在15，它的网站声明yield from，即使在内部生成gen的进步，从一个yield语句到下一个。从外面我们的角度来看caller，如果它产生值的的英文从我们不能告诉caller或从发电机将其委托给。而从内部gen，不能我们当值从发送告知caller或从外面。该yield from语句的英文一个无摩擦信道，通过该进出的值流gen，直到gen完成。

协同一个程序可以工作委托给一个子协程与yield from状语从句：接收工作的查询查询结果。通知，上面，即caller印刷“的产量从返回值：完成”。当gen完成后，报道查看其值成为价值yield from的声明caller：

    rv = yield from gen

早些时候，当我们批评基于回调的异步编程，我们最刺耳的抱怨是关于“堆栈翻录”：当回调抛出一个异常，跟踪堆栈通常的英文没用的那只能说明事件循环正在运行的回调，不是而原因。如何协程票价？

>>> def gen_fn():
...     raise Exception('my error')
>>> caller = caller_fn()
>>> caller.send(None)
Traceback (most recent call last):File "<input>", line 1, in <module>File "<input>", line 3, in caller_fnFile "<input>", line 2, in gen_fn
Exception: my error

这是更为有用！堆栈跟踪显示caller_fn被委托给gen_fn当它扔了错误。更令人欣慰的，我们可以换调用子协同程序中的异常处理程序，同样是与正常子程序：

>>> def gen_fn():
...     yield 1
...     raise Exception('uh oh')
...
>>> def caller_fn():
...     try:
...         yield from gen_fn()
...     except Exception as exc:
...         print('caught {}'.format(exc))
...
>>> caller = caller_fn()
>>> caller.send(None)
1
>>> caller.send('hello')
caught uh oh

因此，我们因子随子协同程序，就像一般的子程序逻辑。让我们因数我们的抓取工具一些有用的子协同程序。我们写一个read协同程序接收一个块：

def read(sock):f = Future()def on_readable():f.set_result(sock.recv(4096))selector.register(sock.fileno(), EVENT_READ, on_readable)chunk = yield f  # Read one chunk.selector.unregister(sock.fileno())return chunk

建立我们在read与read_all接收整条消息协程：

def read_all(sock):response = []# Read whole response.chunk = yield from read(sock)while chunk:response.append(chunk)chunk = yield from read(sock)return b''.join(response)

如果你眯着眼睛以正确的方式，该yield from语句消失，这些看起来像是在做阻塞I / O的常规功能。但事实上，read和read_all是协同程序。从产生read停顿read_all，直到I / O完成。虽然read_all暂停，ASYNCIO的事件循环做其他工作，并等待其他I / O活动; read_all恢复用的结果read下一次循环打勾一旦其活动已准备就绪。

在堆栈的根，fetch叫read_all：

class Fetcher:def fetch(self):# ... connection logic from above, then:sock.send(request.encode('ascii'))self.response = yield from read_all(sock)

不可思议的是，任务类需要任何修改。它驱动外fetch协程刚状语从句：以前一样：

Task(fetcher.fetch())
loop()

当read产生一个未来，接收任务经过它的沟道yield from语句，精确地，如果将来从直接得到fetch。当循环解决了一个未来的任务将它的结果为fetch，和值由接受read，好像就开车任务read直接：

图5.3 - 收费由

为了完善我们的协同程序的实现，我们擦亮了一个三月：的我们使用代码yield时，它等待的未来，但是yield from，当它委托给一个子协程如果使用我们这将的英文更精致yield from每当协程暂停。然后，协程不需要用它等待着什么类型的东西关注自身。

我们需要在Python的发电机和迭代器之间的深对应的优势。推进发电机是，给调用者，同样作为前进迭代器。所以，我们做我们的未来类迭代通过实施特殊的方法：

    # Method on Future class.def __iter__(self):# Tell Task to resume me here.yield selfreturn self.result

的未来__iter__方法的英文产生未来本身就是一个协同程序。现在，当我们更换这样的代码：

# f is a Future.
yield f

......有了这个：

# f is a Future.
yield from f

......结果是一样的！驾驶任务从呼叫接收以后send，当未来解决它发送新的结果返回到协程。

的英文什么的使用优势yield from无处不在？这是为什么不是等待与期货更好yield，并委托给子与协同程序yield from？这是更好，因为现在，一个方法可以自由地改变其实施不影响调用者：它可能是一个返回未来，正常将方法的解析为一个值，可能也。所有游戏的英文协同程序yield from语句状语从句：报道查看值。在这两种情况下，者调用只需要yield from在订单的方法来等待查询查询结果。

亲爱的读者，我们已经达到了我们在ASYNCIO协同程序博览会愉快的结束。我们凝视着发电机的机械，勾勒期货和任务的实现。我们概述了ASYNCIO达到两全其美的：并发I / O比线程更有效，比回调更清晰。当然，真正的ASYNCIO比我们的草图复杂得多。真正的框架解决了零拷贝I / O，公平调度，异常处理，以及其它功能丰富。

要在ASYNCIO用户，与协程编码比你在这里看到的要简单得多。在上面的代码中我们实现了从第一原理协同程序，所以你看到的回调，任务和期货。你甚至看到非阻塞socket和调用select。但是，当涉及的时间来建立与ASYNCIO的应用程序，这一切都不出现在你的代码。正如我们承诺，您现在可以真彩提取网址：

    @asyncio.coroutinedef fetch(self, url):response = yield from self.session.get(url)body = yield from response.read()

满足于这样的论述，我们回到最初的分配：写一个异步网络爬虫，用ASYNCIO。

协调协同程序

我们首先介绍如何，我们希望我们的抓取工具的工作。现在是时候与ASYNCIO协程来实现它。

我们的抓取工具将获取的第一页，分析它的链接，并将其添加到队列中。在此之后它的球迷了整个网站，获取网页兼任。但是，限制客户端和服务器上的负载，我们希望工人运行的一些最大数量，并没有更多的。每当一个工人完成抓取的网页，就应立即拉从队列中下一个环节。我们将通过时间的时候没有足够的工作来绕去，所以一些工人必须暂停。但是，当工人打一个网页丰富的新的联系，那么队列突然增长和任何暂停的工人应该警醒，并得到破解。最后，一旦它的工作就完成了我们的程序必须退出。

试想一下，如果工人线程。我们如何表达履带的算法？我们可以使用同步队列11 Python标准库。每一个项目被放入队列时，队列增加它的“任务”计数。工作线程调用task_done一个项目完成后做事。在主线程块Queue.join，把直到队列中的每个项目由一个匹配task_done的呼叫，然后退出。

协同程序使用完全相同的模式与ASYNCIO队列！首先，我们导入：

try:from asyncio import JoinableQueue as Queue
except ImportError:# In Python 3.5, asyncio.JoinableQueue is# merged into Queue.from asyncio import Queue

我们收集了工人的共享状态履带类，并在其撰写的主要逻辑crawl。方法我们开始crawl在协程状语从句：运行ASYNCIO的事件循环，直到crawl结束：

loop = asyncio.get_event_loop()crawler = crawling.Crawler('http://xkcd.com',max_redirect=10)loop.run_until_complete(crawler.crawl())

履带始于根URL和max_redirect，重定向的数量是愿意跟随撷取任何一个URL。它把对(URL, max_redirect)在队列中。（对于原因，敬请关注。）

class Crawler:def __init__(self, root_url, max_redirect):self.max_tasks = 10self.max_redirect = max_redirectself.q = Queue()self.seen_urls = set()# aiohttp's ClientSession does connection pooling and# HTTP keep-alives for us.self.session = aiohttp.ClientSession(loop=loop)# Put (URL, max_redirect) in the queue.self.q.put((root_url, self.max_redirect))

。队列在未中完成的任务数量现在的英文一个回到我们的主要脚本，启动我们循环事件状语从句：crawl方法：

loop.run_until_complete(crawler.crawl())

该crawl。程协揭开序幕的工人它就像一个主线程：它的块join，直到所有任务完成，而工人在后台运行。

    @asyncio.coroutinedef crawl(self):"""Run the crawler until all work is done."""workers = [asyncio.Task(self.work())for _ in range(self.max_tasks)]# When all work is done, exit.yield from self.q.join()for w in workers:w.cancel()

如果工人线程，我们可能不希望立即开始他们的所有。为了避免产生昂贵的线程，直到可以肯定的是，他们是必要的，一个线程池一般生长需求。但是，协程很便宜，所以我们只需启动允许的最大数量。

。有趣的是，要注意，关闭我们了履带当join未来的解决，工人的任务是活着，但暂停：他们都在等待更多的网址，但无人来。因此，主协程退出之前取消它们。否则，Python的解释器关闭，并呼吁所有对象的析构函数，生活任务哭了出来：

ERROR:asyncio:Task was destroyed but it is pending!

的英文汉语中类似的如何cancel工作的？发电机有我们还没有表现出你的特点。您可以抛出一个异常，从外部进入发电机：

>>> gen = gen_fn()
>>> gen.send(None)  # Start the generator as usual.
1
>>> gen.throw(Exception('error'))
Traceback (most recent call last):File "<input>", line 3, in <module>File "<input>", line 2, in gen_fn
Exception: error

发电机是由恢复throw，但现在养一个例外。如果发电机的调用堆栈没有代码捕获它，除了泡沫备份到顶部。因此，要取消任务的协同程序：

    # Method of Task class.def cancel(self):self.coro.throw(CancelledError)

无论发电机暂停，一些在yield from声明，将恢复并抛出异常。我们在处理任务的取消step方法：

    # Method of Task class.def step(self, future):try:next_future = self.coro.send(future.result)except CancelledError:self.cancelled = Truereturnexcept StopIteration:returnnext_future.add_done_callback(self.step)

现在的任务知道它被取消，所以当它被摧毁它不会怒斥光明的消逝。

一旦crawl取消了工人，它退出。该事件循环看到的是，协程完成（我们将看到更高版本），并且它也退出：

loop.run_until_complete(crawler.crawl())

该crawl。方法包括所有的，我们的主要协同程序必须做的。这是工人协程从队列中获得的URL，获取它们，并这些解析新的联系每个工作运行work独立的协程：

    @asyncio.coroutinedef work(self):while True:url, max_redirect = yield from self.q.get()# Download page and add new links to self.q.yield from self.fetch(url, max_redirect)self.q.task_done()

蟒这个看到所有游戏代码yield from语句，它编译成一个发生器功能。所以crawl，主当协同程序调用self.work十倍，它实际上并没有执行这个方法：它仅创建与该代码引用10发生器的个对象它包装每一个任务。任务接收每个未来发生器所产生的，并通过调用驱动发电机send，每个未来的结果，当未来的解决。因为发电机有自己的堆栈帧，他们独立运行，有独立的局部变量和指令指针。

。坐标工人通过队列的研究员它等待与新的网址：

    url, max_redirect = yield from self.q.get()

的队列get方法本身就是一个协同程序：它会暂停，直到有人把一个项目在排队，然后恢复并返回该项目。

顺便说一句，这是那里的工人将在爬行的最后，当主协同程序取消其暂停。从协程的角度来看，其当周围的循环求最后一趟结束yield from提出了一个CancelledError。

当一个工人获取它解析链接的页面，并在队列中提出新的，调用然后task_done递减计数器位。最终，一名工人取出其URL已全部取出已经一个页面，也没有留在队列中的工作。因此，这名工人的号召，task_done计数器递减至零。然后crawl，正在它等待队列的join方法，取消暂停和完成。

我们承诺要解释为何在队列中的项目是对，如：

# URL to fetch, and the number of redirects left.
('http://xkcd.com/353', 10)

新的网址，有其余10个重定向。在获取重定向这个特定的URL结果到新的位置以斜杠。我们递减剩余重定向的数量，并把下一个位置在队列中：

# URL with a trailing slash. Nine redirects left.
('http://xkcd.com/353/', 9)

在aiohttp我们使用默认会进行重定向，给我们最终的响应包。我们告诉它不要，但是，在履带处理重定向，所以它可以凝聚重定向导致相同的目标路径：如果我们已经看到这个URL，它在的英文self.seen_urls我们已经从一个不同的开始这条道路上入口点：

图5.4 - 重定向

爬虫抓取“富”，并认为它重定向到“巴兹”，所以它增加了“巴兹”队列和seen_urls。如果获取下一页是“酒吧”，这也重定向到“巴兹”中，取出不排队“巴兹”一次。如果响应是一个页面，而不是重定向，fetch分析它的链接，并在队列中提出新的问题。

    @asyncio.coroutinedef fetch(self, url, max_redirect):# Handle redirects ourselves.response = yield from self.session.get(url, allow_redirects=False)try:if is_redirect(response):if max_redirect > 0:next_url = response.headers['location']if next_url in self.seen_urls:# We have been down this path before.return# Remember we have seen this URL.self.seen_urls.add(next_url)# Follow the redirect. One less redirect remains.self.q.put_nowait((next_url, max_redirect - 1))else:links = yield from self.parse_links(response)# Python set-logic:for link in links.difference(self.seen_urls):self.q.put_nowait((link, self.max_redirect))self.seen_urls.update(links)finally:# Return connection to pool.yield from response.release()

如果这是多线程的代码，这将是糟糕的与种族的条件。例如，工作人员检查是否有联系是seen_urls，如果没有工人把它在队列中，并增加了它seen_urls。如果它在两个操作之间中断，那么另一名工人可能解析同一链路从一个不同的页面，也可看到它不是seen_urls，也把它添加到队列中。现在，同样的链接在队列中两次，导致（最好）至重复性工作和错误的统计数据。

然而，程协只在容易受到中断yield from的语句。这是一个关键的区别，使得远不太容易的比赛比多线程代码协同程序代码：多线程代码必须明确进入临界区，通过抓一把锁，否则就中断。一个Python的协同程序默认是不间断的，只有割让控制时，它明确地产生。

我们不再需要取出器类像我们曾在基于回调的方案。这课是回调的缺陷解决方法：他们需要一些地方来存储状态，同时等待I / O，因为他们的局部变量不能跨调用保留。但是，fetch协程可以存储它的状态在局部变量像一个普通的功能呢，所以有一类不再需要。

当fetch完成处理时，它返回给调用者的服务器响应，work。该work方法调用task_done的队列，然后会从队列中下一个URL是牵强。

当fetch在队列中提出新的链接它增加的未完成的任务的数量保持状语从句：主协程，这是等待q.join，暂停。然而，如果没有看不见的联系，这是在队列中最后一个网址，当那么work调用task_done的未完成的任务计数下降到零。这一事件取消暂停join状语从句：主协程完成。

负责协调工人与主协程队列代码是这​​样的13：

class Queue:def __init__(self):self._join_future = Future()self._unfinished_tasks = 0# ... other initialization ...def put_nowait(self, item):self._unfinished_tasks += 1# ... store the item ...def task_done(self):self._unfinished_tasks -= 1if self._unfinished_tasks == 0:self._join_future.set_result(None)@asyncio.coroutinedef join(self):if self._unfinished_tasks > 0:yield from self._join_future

主协程，crawl从产生join。因此，当最后一名工人递减的未完成的任务计数为零，标志着它crawl以恢复状语从句：完成。

乘坐快结束了。我们的计划开始与呼叫crawl：

loop.run_until_complete(self.crawler.crawl())

程序如何结束？由于crawl的英文发电机的功能，调用它返回一个发电机。以驱动发电机，ASYNCIO把它包装在一个任务：

class EventLoop:def run_until_complete(self, coro):"""Run until the coroutine is done."""task = Task(coro)task.add_done_callback(stop_callback)try:self.run_forever()except StopError:passclass StopError(BaseException):"""Raised to stop the event loop."""def stop_callback(future):raise StopError

当任务完成时，它会引发StopError，其循环使用的，它已经到达正常完成的信号。

但是，这是什么？该任务名为方法add_done_callback状语从句：result？你可能会认为一个任务类似于一个未来。你的直觉是正确的。我们必须承认，关于我们向你隐瞒了任务类的细节：一个任务是一个未来。

class Task(Future):"""A coroutine wrapped in a Future."""

正常情况下，的英文未来叫别人解决set_result就可以了。但任务解决自身的协同程序停止时，从我们前面的，当发电机返回时，它抛出的特殊的Python的生成探索记住StopIteration例外：

    # Method of class Task.def step(self, future):try:next_future = self.coro.send(future.result)except CancelledError:self.cancelled = Truereturnexcept StopIteration as exc:# Task resolves itself with coro's return# value.self.set_result(exc.value)returnnext_future.add_done_callback(self.step)

因此，当事件循环调用task.add_done_callback(stop_callback)，它准备通过任务停止。这的英文run_until_complete一次：

    # Method of event loop.def run_until_complete(self, coro):task = Task(coro)task.add_done_callback(stop_callback)try:self.run_forever()except StopError:pass

任务当抓住StopIteration并解决本身的回调引起了StopError从内环路，循环停止，调用堆栈展开来run_until_complete。我们的计划已完成。

结论

越来越多的时候，现代化的计划是I / O限制，而不是CPU限制。对于这样的节目，Python中的线程是最坏的两个世界：全局解释锁防止它们实际执行的计算并行地和抢先切换使得它们容易种族。异步往往是正确的模式。但是，随着基于回调的异步代码的增长，它往往成为一个披头散发的烂摊子。协同程序是一个整洁的选择。他们自然因素为子程序，与健全的异常处理和堆栈跟踪。

如果我们眯着眼睛，这样的yield from语句模糊，协程看起来像一个线程执行传统的阻塞I / O. 我们甚至可以协调与多线程编程的经典模式的协同程序。有没有必要再造。因此，相比于回调，协程是一个迷人的成语与多线程经历的编码器。

但是，我们当打开我们的眼睛状语从句：重点yield from发言，我们看到它们标志着分的时候，协程割让控制，并允许他人经营。不像线程，协程显示在那里我们的代码可以被中断，它不能。在他的启发的文章“硬骨头” 14，莱夫科维茨雕文写道，“线程使局部推理困难，和当地的推理或许是在软件开发中最重要的事情。” 明确收益，但是，能够“通过检查程序本身，而不是检查整个系统了解的行为（从而，正确性）例行的”。

本章在Python和异步历史上的文艺复兴时期写的。基于生成器的协同程序，其制定你刚才了解到，在2014年3月发布的“ASYNCIO”模块中使用Python 3.4在2015年9月，Python的3.5发布内置了语言本身协程。而不是“从收益率”与新的语法“异步高清”，宣布这些本土coroutinesare，并且，他们使用新的“等待”关键字委托给协程或等待未来。

尽管取得了这些进展，核心思想依然存在。Python的新的本地协同程序将从发电机语法不同，但非常相似的工作; 的确，他们将分享Python解释器中的实现。任务，未来和事件循环将继续在ASYNCIO发挥他们的作用。

现在你知道ASYNCIO协同程序是如何工作的，你可以在很大程度上忘记的细节。机器是一个短小精悍的接口后面卷起。但是，你的基本面的把握，使您能够在现代异步环境中正确和有效编码。