初识 asyncio/aiohttp

异步编程并不简单。相比平常的同步编程，你需要付出更多的努力在使用回调函数，以事件以及事件处理器的模式进行思考。同时也是因为asyncio相对较新，相关的教程以及博客还很少的缘故。官方文档非常简陋，只有最基本的范例。在我写本文的时候，Stack Overflow上面，只有410个与asyncio相关的话题（相比之下，twisted相关的有2585）。有个别关于asyncio的不错的博客以及文章，比如这个、这个、这个，或者还有这个
以及这个。
简单起见，我们先从基础开始 —— 简单HTTP hello world —— 发起GET请求，同时获取一个单独的HTTP响应。

同步模式，你这么做：

import requests 
def hello()    
     return requests.get("http://httpbin.org/get")     
print(hello())

接着我们使用aiohttp：

#!/usr/local/bin/python3.5 
import asyncio 
from aiohttp import ClientSession 

async def hello():     
    async with ClientSession() as session:         
        async with session.get("http://httpbin.org/headers") as response:                
            response = await response.read()                         
            print(response) 
  
loop = asyncio.get_event_loop() 
loop.run_until_complete(hello())

好吧，看上去仅仅一个简单的任务，我写了很多的代码……那里有“async def”、“async with”、“await”—— 看上去让人迷惑，让我们尝试弄懂它们。
你使用async以及await关键字将函数异步化。在hello()中实际上有两个异步操作：首先异步获取相应，然后异步读取响应的内容。
Aiohttp推荐使用ClientSession作为主要的接口发起请求。ClientSession允许在多个请求之间保存cookie以及相关对象信息。Session(会话)在使用完毕之后需要关闭，关闭Session是另一个异步操作，所以每次你都需要使用async with关键字。
一旦你建立了客户端session，你可以用它发起请求。这里是又一个异步操作的开始。上下文管理器的with语句可以保证在处理session的时候，总是能正确的关闭它。

要让你的程序正常的跑起来，你需要将他们加入事件循环中。所以你需要创建一个asyncio loop的实例， 然后将任务加入其中。

看起来有些困难，但是只要你花点时间进行思考与理解，就会有所体会，其实并没有那么复杂。

访问多个链接

同步方式如下：

for url in urls:     
    print(requests.get(url).text)

很简单。不过异步方式却没有这么容易。所以任何时候你都需要思考，你的处境是否有必要用到异步。如果你的app在同步模式工作的很好，也许你并不需要将之迁移到异步方式。如果你确实需要异步方式，这里会给你一些启示。我们的异步函数hello()还是保持原样，不过我们需要将之包装在asyncio的Future对象中，然后将Future对象列表作为任务传递给事件循环。

loop = asyncio.get_event_loop() 
tasks = [] # I'm using test server localhost, but you can use any url 
url = "http://localhost:8080/{}" 
for i in range(5):     
    task = asyncio.ensure_future(hello(url.format(i)))     
    tasks.append(task) 

loop.run_until_complete(asyncio.wait(tasks))

现在假设我们想获取所有的响应，并将他们保存在同一个列表中。目前，我们没有保存响应内容，仅仅只是打印了他们。让我们返回他们，将之存储在一个列表当中，最后再打印出来。

为了达到这个目的，我们需要修改一下代码：

#!/usr/local/bin/python3.5 
import asyncio 
from aiohttp import ClientSession 

async def fetch(url):      
   async with ClientSession() as session:           
      async with session.get(url) as response:             
          return await response.read()             
          
async def run(loop,  r):     
   url = "http://localhost:8080/{}"         
   tasks = []         
   for i in range(r):             
       task = asyncio.ensure_future(fetch(url.format(i)))                 
       tasks.append(task)         
       responses = await asyncio.gather(*tasks)         
       # you now have all response bodies in this variable         
       print(responses)     
       
def print_responses(result):       
    print(result)     
    
loop = asyncio.get_event_loop() 
future = asyncio.ensure_future(run(loop, 4)) 
loop.run_until_complete(future)

注意asyncio.gather()的用法，它搜集所有的Future对象，然后等待他们返回。

常见错误

现在我们来模拟真实场景，去调试一些错误，作为演示范例。

# WARNING! BROKEN CODE DO NOT COPY PASTE 
async def fetch(url):      
   async with ClientSession() as session:           
      async with session.get(url) as response:                     
          return response.read()

如果你对aiohttp或者asyncio不够了解，即使你很熟悉Python，这段代码也不好debug。

上面的代码产生如下输出：

pawel@pawel-VPCEH390X ~/p/l/benchmarker> ./bench.py  
[<generator object ClientResponse.read at 0x7fa68d465728>,
 <generator object ClientResponse.read at 0x7fa68cdd9468>,
 <generator object ClientResponse.read at 0x7fa68d4656d0>,
 <generator object ClientResponse.read at 0x7fa68cdd9af0>]

发生了什么？你期待获得响应对象，但是你得到的是一组生成器。怎么会这样？

我之前提到过，response.read()是一个异步操作，这意味着它不会立即返回结果，仅仅返回生成器。这些生成器需要被调用跟运行，但是这并不是默认行为。在Python34中加入的yield from以及Python3.5中加入的await便是为此而生。它们将迭代这些生成器。以上代码只需要在response.read()前加上await关键字即可修复。如下：

    # async operation must be preceded by await          
    return await response.read() 
    # NOT: return response.read()

我们看看另一个例子。

# WARNING! BROKEN CODE DO NOT COPY PASTE 
async def run(loop,  r):       
   url = "http://localhost:8080/{}"         
   tasks = []         
   for i in range(r):             
       task = asyncio.ensure_future(fetch(url.format(i)))                 
       tasks.append(task)         
       responses = asyncio.gather(*tasks)         
       print(responses)

输出结果如下：

pawel@pawel-VPCEH390X ~/p/l/benchmarker> ./bench.py  
<_GatheringFuture pending> 
Task was destroyed but it is pending! 
task: <Task pending coro=<fetch() running at ./bench.py:7> 
        wait_for=<Future pending cb=[Task._wakeup()]> 
        cb=[gather.<locals>._done_callback(0)() 
        at /usr/local/lib/python3.5/asyncio/tasks.py:602]> 
Task was destroyed but it is pending! 
task: <Task pending coro=<fetch() running at ./bench.py:7> 
　　wait_for=<Future pending cb=[Task._wakeup()]> 
　　cb=[gather.<locals>._done_callback(1)() 
　　at /usr/local/lib/python3.5/asyncio/tasks.py:602]> 
Task was destroyed but it is pending! 
task: <Task pending coro=<fetch() running at ./bench.py:7> 
　　wait_for=<Future pending cb=[Task._wakeup()]> 
　　cb=[gather.<locals>._done_callback(2)() 
　　at /usr/local/lib/python3.5/asyncio/tasks.py:602]> 
Task was destroyed but it is pending! 
task: <Task pending coro=<fetch() running at ./bench.py:7>
　　wait_for=<Future pending cb=[Task._wakeup()]> 
　　cb=[gather.<locals>._done_callback(3)() 
　　at /usr/local/lib/python3.5/asyncio/tasks.py:602]>

发生了什么？查看本地日志，你会发现没有任何请求到达服务器，实际上没有任何请求发生。打印信息首先打印<_Gathering pending>对象，然后警告等待的任务被销毁。又一次的，你忘记了await。

修改

    responses = asyncio.gather(*tasks)

到

    responses = await asyncio.gather(*tasks)

即可解决问题。

经验：任何时候，你在等待什么的时候，记得使用await。

参考

同步 vs 异步

看看同步与异步（client）效率上的区别。异步每分钟能够发起多少请求。
为此，我们首先配置一个异步的aiohttp服务器端。这个服务端将获取全部的html文本， 来自Marry Shelley的Frankenstein。在每个响应中，它将添加随机的延时。有的为0，最大值为3s。类似真正的app。有些app的响应延时为固定值，一般而言，每个响应的延时是不同的。

服务器代码如下：

#!/usr/local/bin/python3.5 
import asyncio 
from datetime import datetime 
from aiohttp import web 
import random 

# set seed to ensure async and sync client get same distribution of delay values 
# and tests are fair random.seed(1) 
async def hello(request):     
    name = request.match_info.get("name", "foo")         
    n = datetime.now().isoformat()         
    delay = random.randint(0, 3)         
    await asyncio.sleep(delay)         
    headers = {"content_type": "text/html", "delay": str(delay)}         
    # opening file is not async here, so it may block, to improve         
    # efficiency of this you can consider using asyncio Executors         
    # that will delegate file operation to separate thread or process         
    # and improve performance         
    # https://docs.python.org/3/library/asyncio-eventloop.html#executor         
    # https://pymotw.com/3/asyncio/executors.html     
    with open("frank.html", "rb") as html_body:            
         print("{}: {} delay: {}".format(n, request.path, delay))                 
         response = web.Response(body=html_body.read(), headers=headers)         
         return response     
         
app = web.Application() 
app.router.add_route("GET", "/{name}", hello) 
web.run_app(app)

同步客户端代码如下：

import requests 
r = 100 
url = "http://localhost:8080/{}" 
for i in range(r):      
   res = requests.get(url.format(i))      
  delay = res.headers.get("DELAY")         
  d = res.headers.get("DATE")         
  print("{}:{} delay {}".format(d, res.url, delay))

在我的机器上，上面的代码耗时2分45s。而异步代码只需要3.48s。

有趣的是，异步代码耗时无限接近最长的延时（server的配置）。如果你观察打印信息，你会发现异步客户端的优势有多么巨大。有的响应为0延迟，有的为3s。同步模式下，客户端会阻塞、等待，你的机器什么都不做。异步客户端不会浪费时间，当有延迟发生时，它将去做其他的事情。在日志中，你也会发现这个现象。首先是0延迟的响应，然后当它们到达后，你将看到1s的延迟，最后是最大延迟的响应。

极限测试

现在我们知道异步表现更好，让我们尝试去找到它的极限，同时尝试让它崩溃。我将发送1000异步请求。我很好奇我的客户端能够处理多少数量的请求。

> time python3 bench.py 
2.68user 0.24system 0:07.14elapsed 40%CPU 
(0avgtext+0avgdata 53704maxresident)
k 0inputs+0outputs (0major+14156minor)pagefaults 0swaps

1000个请求，花费了7s。相当不错的成绩。然后10K呢？很不幸，失败了：

responses are <_GatheringFuture finished exception=
　　ClientOSError(24, 'Cannot connect to host localhost:8080 ssl:
　　False [Can not connect to localhost:8080 [Too many open files]]')> 
Traceback (most recent call last):  
   File "/home/pawel/.local/lib/python3.5/site-packages/aiohttp/connector.py", line 581, in _create_connection  
   File "/usr/local/lib/python3.5/asyncio/base_events.py", line 651, in create_connection   
   File "/usr/local/lib/python3.5/asyncio/base_events.py", line 618, in create_connection    
   File "/usr/local/lib/python3.5/socket.py", line 134, in __init__ OS
   Error: [Errno 24] Too many open files</pre>

这样不大好，貌似我倒在了10K connections problem面前。

traceback显示，open files太多了，可能代表着open sockets太多。为什么叫文件？Sockets（套接字）仅仅是文件描述符，操作系统有数量的限制。多少才叫太多呢？我查看Python源码，然后发现这个值为1024.怎么样绕过这个问题？一个粗暴的办法是增加这个数值，但是听起来并不高明。更好的办法是，加入一些同步机制，限制并发数量。于是我在asyncio.Semaphore()中加入最大任务限制为1000.

修改客户端代码如下：

# modified fetch function with semaphore 
import random 
import asyncio 
from aiohttp import ClientSession 

async def fetch(url):      
   async with ClientSession() as session:         
       async with session.get(url) as response:                     
       　delay = response.headers.get("DELAY")                     
       　date = response.headers.get("DATE")                     
       　print("{}:{} with delay {}".format(date, response.url, delay))                     
       　return await response.read() 
       　
async def bound_fetch(sem, url):     
    # getter function with semaphore         
    async with sem:                 
    　await fetch(url) 
    async def run(loop,  r):         
    　url = "http://localhost:8080/{}"         
    　tasks = []         
    　# create instance of Semaphore         
    　sem = asyncio.Semaphore(1000)         
    　for i in range(r):             
    　    # pass Semaphore to every GET request                 
    　    task = asyncio.ensure_future(bound_fetch(sem, url.format(i)))                 
    　    tasks.append(task)         
    　    responses = asyncio.gather(*tasks)         
    　    
await responses number = 10000 
loop = asyncio.get_event_loop() 
future = asyncio.ensure_future(run(loop, number)) 
loop.run_until_complete(future)

现在，我们可以处理10k链接了。这花去我们23s，同时返回了一些异常。不过不管怎样，相当不错的表现。