使用协程构建高性能爬虫
一、简介
在执行一些 IO 密集型任务的时候，程序常常会因为等待 IO 而阻塞。比如在网络爬虫中，如果我们使用 requests 库来进行请求的话，如果网站响应速度过慢，程序一直在等待网站响应，最后导致其爬取效率是非常非常低的。为了解决这类问题，本文就来探讨一下 Python 中异步协程来加速的方法，此种方法对于 IO 密集型任务非常有效。如将其应用到网络爬虫中，爬取效率甚至可以成倍地提升。本文使用 async/await 来实现，需要 Python 3.5 及以上版本。
二、概念介绍
1.阻塞
程序未得到所需计算资源时被挂起的状态。程序在等待某个操作完成期间，自身无法继续干别的事情，则称该程序在该操作上是阻塞的。阻塞形式有：网络 I/O 阻塞、磁盘 I/O 阻塞、用户输入阻塞等。包括 CPU 切换上下文时，多核 CPU 则正在执行上下文切换操作的核不可被利用。
2.非阻塞
程序在等待某操作过程中，自身不被阻塞，可以继续运行干别的事情，则称该程序在该操作上是非阻塞的。非阻塞的存在是因为阻塞存在，正因为某个操作阻塞导致的耗时与效率低下，我们才要把它变成非阻塞，以提高效率。
3.同步
不同程序单元为了完成某个任务，在执行过程中需靠某种通信方式以协调一致，称这些程序单元是同步执行的。例如购物系统中更新商品库存，需要用“锁”作为通信信号，让不同的更新请求强制排队顺序执行，那更新库存的操作是同步的。同步意味着有序。
4.异步
不同程序单元之间过程中无需通信协调，也能完成任务的方式，不相关的程序单元之间可以是异步的。爬虫下载网页。调度程序调用下载程序后，即可调度其他任务，而无需与该下载任务保持通信以协调行为。不同网页的下载、保存等操作都是无关的，也无需相互通知协调。这些异步操作的完成时刻并不确定,异步是无序的。
5.协程
Coroutine，又称微线程，纤程，协程是一种用户态的轻量级线程。拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此协程能保留上一次调用时的状态，即所有局部状态的一个特定组合，每次过程重入时，就相当于进入上一次调用的状态。协程本质上是个单进程，协程相对于多进程来说，无需线程上下文切换的开销，无需原子操作锁定及同步的开销，编程模型也非常简单。
网络爬虫场景下，我们发出一个请求之后，需要等待一定的时间才能得到响应，但其实在这个等待过程中，程序可以干许多其他的事情，等到响应得到之后才切换回来继续处理，这样可以充分利用 CPU 和其他资源，这就是异步协程的优势
6.异步协程asyncio
Python 中使用协程最常用的库莫过于 asyncio。
- event_loop：事件循环，相当于一个无限循环，我们可以把一些函数注册到这个事件循环上，当满足条件发生的时候，就会调用对应的处理方法。
- coroutine：中文翻译叫协程，在 Python 中常指代为协程对象类型，我们可以将协程对象注册到时间循环中，它会被事件循环调用。我们可以使用 async 关键字来定义一个方法，这个方法在调用时不会立即被执行，而是返回一个协程对象。
- task：任务，它是对协程对象的进一步封装，包含了任务的各个状态。
- future：代表将来执行或没有执行的任务的结果，实际上和 task 没有本质区别。
另外我们还需要了解 async/await 关键字，它是从 Python 3.5 才出现的，专门用于定义协程。其中，async 定义一个协程，await 用来挂起阻塞方法的执行。
三、代码实现
1.首先我们使用Flask简单的实现一个服务器
如果没有安装 Flask 的话可以执行如下命令安装：
pip3 install flask
然后编写服务器代码如下：
from flask import Flask import time app = Flask(__name__) @app.route('/') def index(): # 模拟耗时io time.sleep(2) return 'hello' if __name__ == '__main__': # 启动多线程模式 app.run(threaded=True)
这里我们定义了一个 Flask 服务，主入口是 index() 方法，方法里面先调用了 sleep() 方法休眠 2 秒，然后接着再返回结果，也就是说，每次请求这个接口至少要耗时 2 秒，这样我们就模拟了一个慢速的服务接口。
2.使用asyncio进行测试
import asyncio import requests import time start = time.time() async def get(url): return requests.get(url) async def request(): url = 'http://127.0.0.1:5000' print('Waiting for ', url) response = await get(url) print('Get response from ', url, 'Result:', response.text) tasks = [asyncio.ensure_future(request()) for _ in range(5)] loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks)) end = time.time() print( 'Cost time:', end - start )
在这里我们还是创建了五个 task，然后将 task 列表传给 wait() 方法并注册到时间循环中执行
输出结果：
Waiting for http://127.0.0.1:5000 Get response from http://127.0.0.1:5000 Result: hello Waiting for http://127.0.0.1:5000 Get response from http://127.0.0.1:5000 Result: hello Waiting for http://127.0.0.1:5000 Get response from http://127.0.0.1:5000 Result: hello Waiting for http://127.0.0.1:5000 Get response from http://127.0.0.1:5000 Result: hello Waiting for http://127.0.0.1:5000 Get response from http://127.0.0.1:5000 Result: hello Cost time: 10.043976068496704
可以发现和正常的请求并没有什么两样，依然还是顺次执行的，耗时 10 秒，平均一个请求耗时 2 秒，说好的异步处理呢？其实，要实现异步处理，我们得先要有挂起的操作，当一个任务需要等待 IO 结果的时候，可以挂起当前任务，转而去执行其他任务，这样我们才能充分利用好资源，上面方法都是一本正经的串行走下来，连个挂起都没有，怎么可能实现异步？想太多了。要实现异步，接下来我们再了解一下 await 的用法，使用 await 可以将耗时等待的操作挂起，让出控制权。当协程执行的时候遇到 await，时间循环就会将本协程挂起，转而去执行别的协程，直到其他的协程挂起或执行完毕。
仅仅将涉及 IO 操作的代码封装到 async 修饰的方法里面是不可行的！我们必须要使用支持异步操作的请求方式才可以实现真正的异步，所以这里就需要 aiohttp 派上用场了。
3.使用aiohttp
aiohttp 是一个支持异步请求的库，利用它和 asyncio 配合我们可以非常方便地实现异步请求操作。
安装方式如下：
pip3 install aiohttp
官方文档链接为：https://aiohttp.readthedocs.io/，它分为两部分，一部分是 Client，一部分是 Server，详细的内容可以参考官方文档。
import aiohttp import asyncio import time start = time.time() async def get(url): session = aiohttp.ClientSession() # 实例化Clientsession()对象 response = await session.get(url) # 支持get(),post()，params/data,proxy='..'等参数 result = await response.text() # text()字符串，json()json类型，read()二进制 await session.close() # 关闭资源，使用with语句可以自动释放 return result async def request(): url = 'http://127.0.0.1:5000' print('Waiting fro ', url) # result = await get(url) result = await get_w(url) print('Get response from ', url, 'Result:', result) tasks = [asyncio.ensure_future(request()) for _ in range(5)] loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks)) end = time.time() print('Cost time:', end - start)
输出结果如下：
Waiting fro http://127.0.0.1:5000 Waiting fro http://127.0.0.1:5000 Waiting fro http://127.0.0.1:5000 Waiting fro http://127.0.0.1:5000 Waiting fro http://127.0.0.1:5000 Get response from http://127.0.0.1:5000 Result: hello Get response from http://127.0.0.1:5000 Result: hello Get response from http://127.0.0.1:5000 Result: hello Get response from http://127.0.0.1:5000 Result: hello Get response from http://127.0.0.1:5000 Result: hello Cost time: 2.012542963027954
开始运行时，时间循环会运行第一个 task，针对第一个 task 来说，当执行到第一个 await 跟着的 get() 方法时，它被挂起，但这个 get() 方法第一步的执行是非阻塞的，挂起之后立马被唤醒，所以立即又进入执行，创建了 ClientSession 对象，接着遇到了第二个 await，调用了 session.get() 请求方法，然后就被挂起了，由于请求需要耗时很久，所以一直没有被唤醒，好第一个 task 被挂起了，那接下来该怎么办呢？事件循环会寻找当前未被挂起的协程继续执行，于是就转而执行第二个 task 了，也是一样的流程操作，直到执行了第五个 task 的 session.get() 方法之后，全部的 task 都被挂起了。所有 task 都已经处于挂起状态，那咋办？只好等待了。2 秒之后，几个请求几乎同时都有了响应，然后几个 task 也被唤醒接着执行，输出请求结果，最后耗时，2秒！
上面的代码也可以配合with使用，
# 使用with语句 async def get_w(rul): async with aiohttp.ClientSession() as session: async with await session.get(rul) as response: result = await response.text() return result
4.与多进程进行结合使用aiomultiprocess，Python3.6以上版本适用
安装方式：
pip3 install aiomultiprocess
import asyncio import aiohttp import time from aiomultiprocess import Pool start = time.time() async def get(url): session = aiohttp.ClientSession() response = await session.get(url) result = await response.text() session.close() return result async def request(): url = 'http://127.0.0.1:5000' urls = [url for _ in range(100)] async with Pool() as pool: result = await pool.map(get, urls) return result coroutine = request() task = asyncio.ensure_future(coroutine) loop = asyncio.get_event_loop() loop.run_until_complete(task) end = time.time() print('Cost time:', end - start)
当然最后的耗时结果其实和异步是差不多的
做爬取的时候遇到的情况千变万化，一方面我们使用异步协程来防止阻塞，另一方面我们使用 multiprocessing 来利用多核成倍加速，节省时间其实还是非常可观的。