Question

最简单的异步库之一就是gevent。它遵照了让异步函数返回future的模式，意味着代码中的大部分逻辑会保持一样。此外，gevent对标准I/O函数做了猴子补丁，把它们变成了异步，这样大多数时间你可以仅仅使用标准的I/O包并得益于异步的行为。

gevent提供了两个机制来使能异步编程——就如我们刚才提到的，它用异步的I/O函数给标准库打补丁，并且它也有一个greenlet对象能被用于并发执行。greenlet是一种协程，能够被想象成线程（请看第9章对线程的讨论）。无论怎样，所有的greenlets在同一物理线程上运行。那就是说，gevent的调度器在I/O等待期间使用一个事件循环在所有greenlets间来回切换，而不是用多个CPU来运行它们。大多数情况下，gevent通过使用wait函数来设法尽可能透明化地处理事件循环。wait函数将启动一个事件循环，只要有需要就运行着，直到所有的greenlets结束。正因如此，你的大部分gevent代码以串行方式运行。接着，在某点上，你会设置许多greenlets来做并发任务，并且用wait函数来启动事件循环。当wait函数正在执行时，你入队堆积起来的所有并发任务会运行直到结束（或某个停止条件），接着你的代码会重新回到串行方式运行。

Future由gevent.spawn来创建，使用了一个函数和传递给这个函数的参数，并且启动了一个负责运行这个函数的greenlet。greenlet能够被看作一个future，因为你声明的函数一旦运行完成，它的值就会包含在greenlet的value域中。

Python标准模型的补丁会让人更难以控制异步函数得以运行的细节和时间。例如，当正在做异步I/O时，我们想要确认的一件事就是我们没有同时打开太多的文件或者连接。如果我们这样做了，就会让远程服务器过载，或者不得不在太多的操作间做上下文切换，从而减慢进程速度。启动与我们要抓取的URL相同数量的greenlets是没有效率的，我们需要一种机制来限制我们同时处理的HTTP请求。

我们能够通过使用信号量来手动控制并发请求的数量，从而同一时刻只从100个greenlets来做HTTP的get请求。信号量确保了同一时刻只有一定数量的协程能进入上下文模块。作为结果，我们能够启动我们所需的所有greenlets来立即抓取URLs，但只有其中100个将会在同一时刻做出HTTP调用。信号量是一种在各种各样的并行代码流程中使用很多的加锁机制。通过基于各种不同的规则来限制你的代码进程，锁能够帮助你确保程序中各个不同的模块之间不会相互干扰。

现在既然我们设置好了所有的futures，并且已经把加锁机制置入了greenlets的控制流，我们就能够等待直到用gevent.iwait函数开始获取结果为止，那样就会得到一个futures的序列，并遍历准备好的项。反之，我们本可以使用gevent.wait，那会阻塞我们程序的执行直到所有的请求做完为止。

我们经历了把我们的请求分块化的麻烦，而不是把它们立即全都发送出去，因为超载的事件循坏会导致性能降低（这对于所有的异步编程都是真实存在的）。从实验中，我们通常看到在同一时刻100个左右的打开连接是有优化作用的（见图8-3）。如果我们打开更少的连接，我们就还是会在I/O等待期间浪费时间。如果打开更多的连接，我们就会在事件循环中太频繁地做上下文切换，给我们的程序增加不必要的负担。那就是说，100这个值取决于许多事情——代码正运行于其上的计算机，事件循环的实现，远端主机的属性，远端主机的期望响应时间等。我们建议在决定选择之前做一些实验。例8-4显示了我们HTTP爬虫的gevent版本代码。

图8-3　寻找并发请求的合适数量

例8-4　gevent HTTP爬虫

from gevent import monkey
monkey.patch_socket()

import gevent
from gevent.coros import Semaphore
import urllib2
import string
import random

def generate_urls(base_url, num_urls):
  for i in xrange(num_urls):
    yield base_url + "".join(random.sample(string.ascii_lowercase, 10))

def chunked_requests(urls, chunk_size=100):
  semaphore = Semaphore(chunk_size) # ❶
  requests = [gevent.spawn(download, u, semaphore) for u in urls] # ❷
  for response in gevent.iwait(requests):
    yield response

def download(url, semaphore):
  with semaphore: # ❸
    data = urllib2.urlopen(url)
    return data.read()

def run_experiment(base_url, num_iter=500):
  urls = generate_urls(base_url, num_iter)
  response_futures = chunked_requests(urls, 100) # ❹
  response_size = sum(len(r.value) for r in response_futures)
  return response_size

if __name__ == "__main__":
  import time
  delay = 100
  num_iter = 500
  base_url = "http://127.0.0.1:8080/add?name=gevent&delay={}&".format(delay)

  start = time.time()
  result = run_experiment(base_url, num_iter)
  end = time.time()
  print("Result: {}, Time: {}".format(result, end - start))

❶ 这里我们生成了一个信号量来让chunk_size下载发生。

❷ 通过把信号量用作一个上下文管理器，我们确保了只有chunk_size数量的greenlets能够在同一时刻运行上下文的主体部分。

❸ 我们能够把所需数量的greenlets放在队列中，知道它们之中没有一个会运行直到我们用wait或iwait启动一个事件循环为止。

❹ response_futures现在持有一个处于完成状态的futures的迭代器，所有这些futures的.value属性中都具有我们所期望的数据。

作为替换，我们能够使用grequests来大大简化我们的gevent代码。尽管gevent提供了所有类型的低级并发socket操作，grequests组合了HTTP库请求和gevent，其结果就是具有很简单的API来做并发HTTP请求（甚至为我们处理信号量逻辑）。使用qrequests，我们的代码变得简单很多，更容易理解，可维护性更好，然而却还是获得了与更低层的gevent代码相提并论的速度提升（见例8-5）。

例8-5　grequests HTTP爬虫

import grequests

def run_experiment(base_url, num_iter=500):
  urls = generate_urls(base_url, num_iter)
  response_futures = (grequests.get(u) for u in urls) # ❶
  responses = grequests.imap(response_futures, size = 100) # ❷
  response_size = sum(len(r.text) for r in responses)
  return response_size

❶ 首先我们创建了请求并得到future。我们选择了把它当作生成器（generator）来做，这样以后我们只需要做与我们准备发出的请求相同次数的估算。

❷ 现在我们能够取得future对象，并把它们映射成真实的响应对象。.imap函数给我们一个生成器（generator）来产生响应对象，我们就是从响应对象来获取数据。

一件重要的事情要引起注意，那就是我们已经使用了gevent和grequests来生成异步的I/O请求，但是我们在I/O等待期间没有做任何非I/O的计算。图8-4显示了我们取得的巨大速度提升。通过在等待前面的请求完成之际发起更多的请求，我们能够取得69倍的速度提升！通过用水平线代表相互之间的请求栈的方式， 我们能够明显看到在前面的请求完成之前新的请求如何正在发送出去。这与串行爬虫（图8-2）的例子形成了鲜明的对比，在串行爬虫的图中，一个线条只有在前面的线条完成之时才开始。而且，我们能够看到更有趣的效果伴随着gevent请求时间线的形状。例如，在大约前100个请求处，我们看到一个停顿，没有发起新请求。这是因为这时我们的信号量第一次命中，我们能够在任何前面的请求完成之前给信号量加锁。在这之后，信号量进入一个平衡态，只有当另一个请求完成时才解锁，并给当前新请求加锁。

图8-4　例8-5的HTTP请求时间表