第6章异步爬虫

　　我们知道爬虫是 IO 密集型任务,例如使用 requests 库来爬取某个站点,当发出一个请求后,程序必须等待网站返回响应,才能接着运行,而在等待响应的过程中,整个爬虫程序是一直在等待的, 实际上没有做任何事情。对于这种情况,我们有没有优化方案呢?

　　当然有,本章我们就来了解一下异步爬虫的基本概念和实现。

6.1 协程的基本原理

　　要实现异步机制的爬虫,那自然和协程脱不了关系。

1. 案例引入

　　在介绍协程之前,先来看一个案例网站,地址为https://www.httpbin.org/delay/5,访问这个链接需要先等待五秒才能得到结果,这是因为服务器强制等待了5 秒时间才返回响应。

　　平时我们浏览网页的时候,绝大部分网页的响应速度还是很快的,如果写爬虫来爬取,那么从发出请求到接收响应的时间不会很长,因此需要我们等待的时间并不多。

　　然而像上面这个网站,发出一次请求至少需要5 秒才能得到响应,如果用 requests 库写爬虫来爬取,那么每次都要等待5 秒及以上才能拿到结果。

　　下面来测试一下,我们用requests 写一个遍历程序,直接遍历100 次案例网站,试试看有什么效果,实现代码如下:

import requests
import logging
import time

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s: %(message)s')

TOTAL_NUMBER = 100
URL = 'https://www.httpbin.org/delay/5'

start_time = time.time()
for _ in range(1, TOTAL_NUMBER + 1):
    logging.info('scraping %s', URL)
    response = requests.get(URL)
end_time = time.time()
logging.info('total time %s seconds', end_time - start_time)

　　这里我们直接用循环的方式构造了100 个请求,使用的是 requests 单线程,在爬取之前和爬取之后分别记录了时间,最后输出了爬取100 个页面消耗的总时间。

　　运行结果如下:

2020-08-03 01:01:36,781 - INFO: scraping https://www.httpbin.org/delay/5
2020-08-03 01:01:43,410 - INFO: scraping https://www.httpbin.org/delay/5

2020-08-03 01:01:50,029 - INFO: scraping https://www.httpbin.org/delay/5
2020-08-03 01:01:56,702 - INFO: scraping https://www.httpbin.org/delay/5
2020-08-03 01:02:03,345 - INFO: scraping https://www.httpbin.org/delay/5
2020-08-03 01:02:09,958 - INFO: scraping https://www.httpbin.org/delay/5
2020-08-03 01:02:16,500 - INFO: scraping https://www.httpbin.org/delay/5
2020-08-03 01:02:23,143 - INFO: scraping https://www.httpbin.org/delay/5
...
2020-08-03 01:12:19,867 - INFO: scraping https://www.httpbin.org/delay/5
2020-08-03 01:12:26,479 - INFO: scraping https://www.httpbin.org/delay/5
2020-08-03 01:12:33,083 - INFO: scraping https://www.httpbin.org/delay/5
2020-08-03 01:12:39,758 - INFO: total time 662.9764430522919 seconds

由于每个页面都至少要等待5 秒才能加载出来,因此100 个页面至少要花费500 秒时间,加上网站本身的负载问题,总的爬取时间最终约为663 秒,大约11 分钟。这在实际情况中是很常见的,有些网站本身加载速度就比较慢,稍慢的可能1~3 秒,更慢的说不定 10 秒以上。如果我们就用 requests 单线程这么爬取,总耗时将会非常大。此时要是打开多线程或多进程来爬取,其爬取速度确实会成倍提升,那么是否有更好的解决方案呢? 本节就来了解一下使用协程实现加速的方法,这种方法对IO 密集型任务非常有效。如果将其应用到网络爬虫中,那么爬取效率甚至可以提升成百倍。

2. 基础知识

了解协程需要先了解一些基础概念,如阻塞和非阻塞、同步和异步、多进程和协程。

阻塞

阻塞状态指程序未得到所需计算资源时被挂起的状态。程序在等待某个操作完成期间,自身无法继续干别的事情,则称该程序在该操作上是阻塞的。常见的阻塞形式有:网络 I/O 阻塞、磁盘 I/O 阻塞、用户输入阻塞等。阻塞是无处不在的,包括在CPU 切换上下文时,所有进程都无法真正干事情,它们也会被阻塞。在多核CPU 的情况下,正在执行上下文切换操作的核不可被利用。

非阻塞

程序在等待某操作的过程中,自身不被阻塞,可以继续干别的事情,则称该程序在该操作上是非阻塞的。非阻塞并不是在任何程序级别、任何情况下都存在的。仅当程序封装的级别可以囊括独立的子程序单元时,程序才可能存在非阻塞状态。非阻塞因阻塞的存在而存在,正因为阻塞导致程序运行的耗时增加与效率低下,我们才要把它变成非阻塞的。

同步

不同程序单元为了共同完成某个任务,在执行过程中需要靠某种通信方式保持协调一致,此时这些程序单元是同步执行的。例如在购物系统中更新商品库存时,需要用“行锁”作为通信信号,强制让不同的更新请求排队并按顺序执行,这里的更新库存操作就是同步的。简言之,同步意味着有序。

异步

为了完成某个任务,有时不同程序单元之间无须通信协调也能完成任务,此时不相关的程序单元

之间可以是异步的。例如,爬取下载网页。调度程序调用下载程序后,即可调度其他任务,无须与该下载任务保持通信以协调行为。不同网页的下载、保存等操作都是无关的,也无须相互通知协调。这些异步操作的完成时刻并不确定。简言之,异步意味着无序。

多进程

多进程就是利用CPU 的多核优势,在同一时间并行执行多个任务,可以大大提高执行效率。

协程

协程,英文叫作 coroutine,又称微线程、纤程,是一种运行在用户态的轻量级线程。协程拥有自己的寄存器上下文和栈。协程在调度切换时,将寄存器上下文和栈保存到其他地方, 等切回来的时候,再恢复先前保存的寄存器上下文和栈。因此,协程能保留上一次调用时的状态, 即所有局部状态的一个特定组合,每次过程重入,就相当于进入上一次调用的状态。协程本质上是个单进程,相对于多进程来说,它没有线程上下文切换的开销,没有原子操作锁定及同步的开销,编程模型也非常简单。我们可以使用协程来实现异步操作,例如在网络爬虫场景下,我们发出一个请求之后,需要等待一定时间才能得到响应,但其实在这个等待过程中,程序可以干许多其他事情,等得到响应之后再切换回来继续处理,这样可以充分利用 CPU 和其他资源,这就是协程的优势。

3. 协程的用法

接下来,我们了解一下协程的实现。从Python 3.4开始,Python 中加入了协程的概念,但这个版本的协程还是以生成器对象为基础。Python 3.5中增加了 async、await,使得协程的实现更为方便。 Python 中使用协程最常用的库莫过于 asyncio,所以本节会以它为基础来介绍协程的用法。首先,需要了解下面几个概念。

event_loop:事件循环,相当于一个无限循环,我们可以把一些函数注册到这个事件循环上,当满足发生条件的时候,就调用对应的处理方法。

coroutine:中文翻译叫协程,在 Python 中常指代协程对象类型,我们可以将协程对象注册到事件循环中,它会被事件循环调用。我们可以使用async 关键字来定义一个方法,这个方法在调用时不会立即被执行,而是会返回一个协程对象。

task:任务,这是对协程对象的进一步封装,包含协程对象的各个状态。

future:代表将来执行或者没有执行的任务的结果,实际上和 task 没有本质区别。另外,我们还需要了解 async、await 关键字,它们是从Python 3.5才开始出现的,专门用于定义协程。其中,前者用来定义一个协程,后者用来挂起阻塞方法的执行。

4. 准备工作

在本节开始之前,请确保安装的Python 版本为3.5及以上,如果版本是3.4及以下,则下方的案例是不能运行的。具体的安装方法可以参考:https://setup.scrape.center/python。安装好合适的Python 版本之后我们就可以开始本节的学习了。

5. 定义协程

我们来定义一个协程,体验一下它和普通进程在实现上的不同之处,代码如下:

import asyncio

async def execute(x):
    print('Number:', x)

coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')

loop = asyncio.get_event_loop()
loop.run_until_complete(coroutine)
print('After calling loop')

运行结果如下:

Coroutine: <coroutine object execute at 0x1034cf830>
After calling execute
Number: 1
After calling loop

　　首先,我们引入了 asyncio 包,这样才可以使用 async 和 await 关键字。然后使用 async 定义了一个 execute 方法,该方法接收一个数字参数 x,执行之后会打印这个数字。随后我们直接调用了 execute 方法,然而这个方法并没有执行,而是返回了一个 coroutine 协程对象。之后我们使用 get_event_loop 方法创建了一个事件循环 loop,并调用 loop 对象的 run_until_complete 方法将协程对象注册到了事件循环中,接着启动。最后,我们才看到 execute 方法打印出了接收的数字。可见,async 定义的方法会变成一个无法直接执行的协程对象,必须将此对象注册到事件循环中才可以执行。前面我们还提到了 task,它是对协程对象的进一步封装,比协程对象多了运行状态,例如 running、 finished 等,我们可以利用这些状态获取协程对象的执行情况。在上面的例子中,当我们把协程对象 coroutine 传递给 run_until_complete 方法的时候,实际上它进行了一个操作,就是将 coroutine 封装成 task 对象。对此,我们也可以显式地进行声明,代码如下所示:

import asyncio

async def execute(x):
    print('Number:', x)
    return x

coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')

loop = asyncio.get_event_loop()
task = loop.create_task(coroutine)
print('Task:', task)
loop.run_until_complete(task)
print('Task:', task)
print('After calling loop')

运行结果如下:

Coroutine: <coroutine object execute at 0x10e0f7830>
After calling execute
Task: <Task pending coro=<execute() running at demo.py:4>>
Number: 1
Task: <Task finished coro=<execute() done, defined at demo.py:4> result=1>
After calling loop

这里我们定义了loop 对象之后,紧接着调用了它的create_task 方法,将协程对象转化为 task 对象,随后打印输出一下,发现它处于pending 状态。然后将 task 对象添加到事件循环中执行,并再次打印出 task 对象,发现它的状态变成了finished,同时还可以看到其 result 变成了1,也就是我们定义的 execute 方法的返回结果。定义 task 对象还有另外一种方式,就是直接调用 asyncio 包的 ensure_future 方法,返回结果也是 task 对象,这样的话我们就可以不借助loop 对象。即使还没有声明loop,也可以提前定义好 task 对象,这种方式的写法如下:

import asyncio

async def execute(x):
    print('Number:', x)
    return x

coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')
task = asyncio.ensure_future(coroutine)
print('Task:', task)
loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:', task)
print('After calling loop')

运行结果如下:

Coroutine: <coroutine object execute at 0x10aa33830>
After calling execute
Task: <Task pending coro=<execute() running at demo.py:4>>
Number: 1
Task: <Task finished coro=<execute() done, defined at demo.py:4> result=1>
After calling loop

可以发现,运行效果都是一样的。

6. 绑定回调

我们也可以为某个 task 对象绑定一个回调方法。来看下面这个例子:

import asyncio
import requests

async def request():
    url = 'https://www.baidu.com'
    status = requests.get(url)
    return status

def callback(task):
    print('Status:', task.result())

coroutine = request()
task = asyncio.ensure_future(coroutine)
task.add_done_callback(callback)
print('Task:', task)
loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:', task)

这里我们定义了 request 方法,在这个方法里请求了百度,并获取其状态码,但是没有编写任何 print 语句。随后我们定义了 callback 方法,这个方法接收一个参数,参数是 task 对象,在这个方法中调用 print 方法打印出了 task 对象的结果。这样就定义好了一个协程对象和一个回调方法。

我们现在希望达到的效果是,当协程对象执行完毕之后,就去执行声明的callback 方法。那么两者怎样关联起来呢?很简单,只要调用add_done_callback 方法就行。我们将 callback 方法传递给封装好的task 对象,这样当task 执行完毕之后,就可以调用callback 方法了。同时 task 对象还会作为参数传递给 callback 方法,调用task 对象的 result 方法就可以获取返回结果了。运行结果如下:

Task: <Task pending coro=<request() running at demo.py:5> cb=[callback() at demo.py:11]>
Status: <Response [200]>
Task: <Task finished coro=<request() done, defined at demo.py:5> result=<Response [200]>>

实际上,即使不使用回调方法,在task 运行完毕之后,也可以直接调用 result 方法获取结果, 代码如下所示:

import asyncio
import requests

async def request():
    url = 'https://www.baidu.com'
    status = requests.get(url)
    return status

coroutine = request()
task = asyncio.ensure_future(coroutine)
print('Task:', task)
loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:', task)
print('Task Result:', task.result())

运行结果是一样的:

Task: <Task pending coro=<request() running at demo.py:4>>
Task: <Task finished coro=<request() done, defined at demo.py:4> result=<Response [200]>>
Task Result: <Response [200]>

7. 多任务协程

在上面的例子中,我们都只执行了一次请求,如果想执行多次请求,应该怎么办呢?可以定义一个 task 列表,然后使用 asyncio 包中的 wait 方法执行。看下面的例子:

import asyncio
import requests

async def request():
    url = 'https://www.baidu.com'
    status = requests.get(url)
    return status

tasks = [asyncio.ensure_future(request()) for _ in range(5)]
print('Tasks:', tasks)
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
for task in tasks:
    print('Task Result:', task.result())

这里我们使用一个 for 循环创建了 5 个 task,它们组成一个列表,然后把这个列表首先传递给 asyncio 包的 wait 方法,再将其注册到事件循环中,就可以发起5 个任务了。最后,输出任务的执行结果,具体如下:

Tasks: [<Task pending coro=<request() running at demo.py:5>>, <Task pending coro=<request() running at
demo.py:5>>, <Task pending coro=<request() running at demo.py:5>>, <Task pending coro=<request() running at
demo.py:5>>, <Task pending coro=<request() running at demo.py:5>>]
Task Result: <Response [200]>
Task Result: <Response [200]>
Task Result: <Response [200]>
Task Result: <Response [200]>
Task Result: <Response [200]>

可以看到,5 个任务被顺次执行,并得到了执行结果。

8. 协程实现

前面说了好一通,又是 async 关键字,又是coroutine,又是task,又是callback的,似乎并没有从中看出协程的优势,反而写法上更加奇怪和麻烦了?别急,上述案例只是为后面的使用作铺垫。接下来,我们正式看看协程在解决IO 密集型任务方面到底有怎样的优势。在前面的代码中,我们用一个网络请求作为例子,这本身就是一个耗时等待操作,因为在请求网页之后需要等待页面响应并返回结果。耗时等待操作一般都是IO 操作,例如文件读取、网络请求等。协程在处理这种操作时是有很大优势的,当遇到需要等待的情况时,程序可以暂时挂起,转而执行其他操作,从而避免因一直等待一个程序而耗费过多的时间,能够充分利用资源。为了表现协程的优势,我们还是以本节开头介绍的网站 https://www.httpbin.org/delay/5为例,因为该网站响应比较慢,所以可以通过爬取时间让大家直观感受到爬取速度的提升。为了让大家更好地理解协程的正确使用方法,这里先来看看大家使用协程时常犯的错误,后面再给出正确的例子作为对比。首先,还是拿之前的 requests 库进行网页请求,之后再重新使用上面的方法请求一遍:

import asyncio
import requests
import time

start = time.time()

async def request():
    url = 'https://www.httpbin.org/delay/5'
    print('Waiting for', url)
    response = requests.get(url)
    print('Get response from', url, 'response', response)

tasks = [asyncio.ensure_future(request()) for _ in range(10)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

end = time.time()
print('Cost time:', end - start)

这里我们还是创建了10 个 task,然后将 task 列表传给 wait 方法并注册到事件循环中执行。运行结果如下:

Waiting for https://www.httpbin.org/delay/5
Get response from https://www.httpbin.org/delay/5 response <Response [200]>
Waiting for https://www.httpbin.org/delay/5
...
Get response from https://www.httpbin.org/delay/5 response <Response [200]>
Waiting for https://www.httpbin.org/delay/5
Get response from https://www.httpbin.org/delay/5 response <Response [200]>
Waiting for https://www.httpbin.org/delay/5
Get response from https://www.httpbin.org/delay/5 response <Response [200]>
Cost time: 66.64284420013428

可以发现,这和正常的请求并没有什么区别,各个任务依然是顺次执行的,耗时66 秒,平均一个请求耗时 6.6 秒,说好的异步处理呢? 其实,要实现异步处理,先得有挂起操作,当一个任务需要等待IO结果的时候,可以挂起当前任务,转而执行其他任务,这样才能充分利用好资源。上面的方法都是一本正经地串行执行下来,连个挂起都没有,怎么可能实现异步?莫不是想太多了。要实现异步,我们再了解一下 await 关键字的用法,它可以将耗时等待的操作挂起,让出控制权。如果协程在执行的时候遇到 await,事件循环就会将本协程挂起,转而执行别的协程,直到其他协程挂起或执行完毕。所以,我们可能会将代码中的request 方法改成如下这样:

async def request():
    url = 'https://www.httpbin.org/delay/5'
    print('Waiting for', url)
    response = await requests.get(url)
    print('Get response from', url, 'response', response)

仅仅是在 requests 前面加了一个关键字 await。然而此时执行代码,会得到如下报错信息:

Waiting for https://www.httpbin.org/delay/5
Waiting for https://www.httpbin.org/delay/5
Waiting for https://www.httpbin.org/delay/5
Waiting for https://www.httpbin.org/delay/5
Task exception was never retrieved
future: <Task finished coro=<request() done, defined at demo.py:8> exception=TypeError("object Response can't
be used in 'await' expression")>
Traceback (most recent call last):
File "demo.py", line 11, in request
response = await requests.get(url)
TypeError: object Response can't be used in 'await' expression

这次协程遇到 await 时确实挂起了,也等待了,但是最后却报出以上错误信息。这个错误的意思是 requests 返回的Response 对象不能和 await一起使用,为什么呢?因为根据官方文档说明,await 后面的对象必须是如下格式之一:

一个原生协程对象;

一个由 types.coroutine 修饰的生成器,这个生成器可以返回协程对象;

由一个包含 __await__ 方法的对象返回的一个迭代器。这里 requests 返回的 Response 对象以上三种格式都不符合,因此报出了上面的错误。有的读者可能已经发现,既然 await 后面可以跟一个协程对象,那么 async 把请求的方法改成协程对象不就可以了吗?于是就代码被改写成如下的样子:

import asyncio
import requests
import time

start = time.time()

async def get(url):
    return requests.get(url)

async def request():
    url = 'https://www.httpbin.org/delay/5'
    print('Waiting for', url)
    response = await get(url)
    print('Get response from', url, 'response', response)

tasks = [asyncio.ensure_future(request()) for _ in range(10)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
end = time.time()
print('Cost time:', end - start)

这里将请求页面的方法独立出来,并用async修饰,就得到了一个协程对象。运行一下看看:

Waiting for https://www.httpbin.org/delay/5
Get response from https://www.httpbin.org/delay/5 response <Response [200]>
Waiting for https://www.httpbin.org/delay/5
Get response from https://www.httpbin.org/delay/5 response <Response [200]>
Waiting for https://www.httpbin.org/delay/5
Get response from https://www.httpbin.org/delay/5 response <Response [200]>
Waiting for https://www.httpbin.org/delay/5
Get response from https://www.httpbin.org/delay/5 response <Response [200]>
Waiting for https://www.httpbin.org/delay/5
Get response from https://www.httpbin.org/delay/5 response <Response [200]>
Cost time: 65.394437756259273

　　还是报错,协程还不是异步执行的,也就是说我们仅仅将涉及 IO 操作的代码封装到 async 修饰的方法里是不可行的。只有使用支持异步操作的请求方式才可以实现真正的异步,这里 aiohttp 就派上用场了。

9. 使用 aiohttp

aiohttp 是一个支持异步请求的库,它和 asyncio 配合使用,可以使我们非常方便地实现异步请求操作。我们使用 pip3 安装即可:

pip3 install aiohttp

　　具体的安装方法可以参考:https://setup.scrape.center/aiohttp。 aiohttp 的官方文档链接为 https://aiohttp.readthedocs.io/,它分为两部分,一部分是 Client,一部分是 Server。下面我们将 aiohttp 投入使用,将代码改写成如下样子:

import asyncio
import aiohttp
import time

start = time.time()

async def get(url):
    session = aiohttp.ClientSession()
    response = await session.get(url)
    await response.text()
    await session.close()
    return response

async def request():
    url = 'https://www.httpbin.org/delay/5'
    print('Waiting for', url)
    response = await get(url)
    print('Get response from', url, 'response', response)

tasks = [asyncio.ensure_future(request()) for _ in range(10)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

end = time.time()
print('Cost time:', end - start)

这里将请求库由 requests 改成了 aiohttp,利用aiohttp 库里 ClientSession 类的 get 方法进行请求, 返回结果如下:

Waiting for https://www.httpbin.org/delay/5
Waiting for https://www.httpbin.org/delay/5
Waiting for https://www.httpbin.org/delay/5
Waiting for https://www.httpbin.org/delay/5
...
Get response from https://www.httpbin.org/delay/5 response <ClientResponse(https://www.httpbin.org/delay/5)
[200 OK]><CIMultiDictProxy('Date': 'Sun, 09 Aug 2020 14:30:22 GMT', 'Content-Type': 'application/json',
'Content-Length': '360', 'Connection': 'keep-alive', 'Server': 'gunicorn/19.9.0',
'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true')>
...
Get response from https://www.httpbin.org/delay/5 response <ClientResponse(https://www.httpbin.org/delay/5)
[200 OK]><CIMultiDictProxy('Date': 'Sun, 09 Aug 2020 14:30:22 GMT', 'Content-Type': 'application/json',
'Content-Length': '360', 'Connection': 'keep-alive', 'Server': 'gunicorn/19.9.0',
'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true')>
Cost time: 6.033240079879761

成功了!我们发现这次请求的耗时直接由51 秒变成了6 秒,耗费时间减少了非常多。这里我们使用了 await,其后面跟着 get 方法。在执行10 个协程的时候,如果遇到 await,就会将当前协程挂起,转而执行其他协程,直到其他协程也挂起或执行完毕,再执行下一个协程。开始运行时,事件循环会运行第一个 task。对于第一个 task 来说,当执行到第一个 await 跟着的 get 方法时,它会被挂起,但这个 get 方法第一步的执行是非阻塞的,挂起之后会立马被唤醒, 立即又进入执行,并创建了ClientSession 对象。接着遇到第二个 await,调用 session.get 请求方法,然后就被挂起了。由于请求需要耗时很久,所以一直没有被唤醒,好在第一个 task 被挂起了, 那么接下来该怎么办呢?事件循环会寻找当前未被挂起的协程继续执行,于是转而去执行第二个 task,流程操作和第一个 task 也是一样的,以此类推,直到执行第十个 task 的session.get 方法之后,全部的 task 都被挂起了。所有 task 都已经处于挂起状态,那怎么办?只好等待了。5 秒之后,几个请求几乎同时有了响应,然后几个 task 也被唤醒接着执行,并输出请求结果,最后总耗时是6 秒! 怎么样?这就是异步操作的便捷之处,当遇到阻塞式操作时,task 被挂起,程序接着去执行其他 task,而不是傻傻地等着,这样可以充分利用 CPU,而不必把时间浪费在等待 IO 上。有人会说,在上面的例子中,发出网络请求后,接下来的5 秒都是在等待,这5 秒之内,CPU 可以处理的 task 数量远不止这些,既然这样的话,那么我们放10 个、20 个、50 个、100 个、1000 个 task 一起执行,最后得到所有结果的耗时不都是差不多的吗?因为这些任务被挂起后都是一起等待的。从理论上来说,确实是这样,不过有个前提,就是服务器即使在同一时刻接收无限次请求,依然要能保证正常返回结果,也就是服务器应该无限抗压,另外还要忽略 IO 传输时延。满足了这两点, 确实可以做到无限个 task 一起执行,并且在预想时间内得到结果。但由于不同服务器处理 task 的实现机制不同,可能某些服务器并不能承受那么高的并发量,因此响应速度也会减慢。这里我们以百度为例,测试一下并发量分别为1、3、5、10、...、500时的耗时情况,代码如下:

import asyncio
import aiohttp
import time

def test(number):
    start = time.time()

    async def get(url):
        session = aiohttp.ClientSession()
        response = await session.get(url)
        await response.text()
        await session.close()
        return response

    async def request():
        url = 'https://www.baidu.com/'
        await get(url)

    tasks = [asyncio.ensure_future(request()) for _ in range(number)]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

    end = time.time()
    print('Number:', number, 'Cost time:', end - start)

for number in [1, 3, 5, 10, 15, 30, 50, 75, 100, 200, 500]:
    test(number)

运行结果如下:

Number: 1 Cost time: 0.05885505676269531
Number: 3 Cost time: 0.05773782730102539
Number: 5 Cost time: 0.05768704414367676
Number: 10 Cost time: 0.15174412727355957
Number: 15 Cost time: 0.09603095054626465
Number: 30 Cost time: 0.17843103408813477
Number: 50 Cost time: 0.3741800785064697
Number: 75 Cost time: 0.2894289493560791
Number: 100 Cost time: 0.6185381412506104
Number: 200 Cost time: 1.0894129276275635
Number: 500 Cost time: 1.8213098049163818

可以看到,在服务器能够承受高并发的前提下,即使我们增加了并发量,其爬取速度也几乎不会太受影响。综上所述,使用了异步请求之后,我们几乎可以在相同时间内实现成百上千倍次的网络请求,把这个运用在爬虫中,速度提升可谓非常可观。

10. 总结

以上便是 Python 中协程的基本原理和用法,在接下来的6.2节中,我们会详细介绍 aiohttp 库的用法和爬取实战,实现快速、高并发的爬取。本节代码参见:https://github.com/Python3WebSpider/AsyncTest。

6.2 aiohttp 的使用

在6.1节,我们介绍了异步爬虫的基本原理和 asyncio 的基本用法,并且在最后简单提及了使用 aiohttp 实现网页爬取的过程。本节我们介绍一下aiohttp 的常见用法。

1. 基本介绍

前面介绍的 asyncio 模块,其内部实现了对TCP、UDP、SSL协议的异步操作,但是对于HTTP请求来说,就需要用aiohttp 实现了。 aiohttp 是一个基于 asyncio 的异步HTTP网络模块,它既提供了服务端,又提供了客户端。其中, 我们用服务端可以搭建一个支持异步处理的服务器,这个服务器就是用来处理请求并返回响应的,类似于 Django、Flask、Tornado 等一些Web服务器。而客户端可以用来发起请求,类似于使用 requests 发起一个HTTP 请求然后获得响应,但requests 发起的是同步的网络请求,aiohttp则是异步的。

本节我们主要了解一下 aiohttp 客户端部分的用法。

2. 基本实例

　　我们来看一个基本的 aiohttp 请求案例,代码如下:

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text(), response.status

async def main():
    async with aiohttp.ClientSession() as session:
        html, status = await fetch(session, 'https://cuiqingcai.com')
        print(f'html: {html[:100]}...')
        print(f'status: {status}')

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

　　这里使用 aiohttp 爬取了我的个人博客,获得了源码和响应状态码,并打印出来,运行结果如下:

html: <!DOCTYPE HTML>
<html>
<head>
<meta charset="UTF-8">
<meta name="baidu-tc-verification" content=...
status: 200

　　由于网页源码过长,这里只截取了输出的一部分。可以看到,我们成功获取了网页的源代码及响应状态码 200,也就是完成了一次基本的HTTP 请求,即我们成功使用aiohttp 通过异步的方式完成了网页爬取。当然,这个操作用之前讲的 requests 也可以做到。

　　能够发现,aiohttp 的请求方法的定义和之前有明显的区别,主要包括如下几点。

首先在导入库的时候,除了必须引入 aiohttp 这个库,还必须引入asyncio 库。因为要实现异步爬取,需要启动协程,而协程则需要借助于 asyncio 里面的事件循环才能执行。除了事件循环, asyncio 里面也提供了很多基础的异步操作。

异步爬取方法的定义和之前有所不同,每个异步方法的前面都要统一加 async 来修饰。

with as 语句前面同样需要加 async 来修饰。在 Python 中, with as 语句用于声明一个上下文管理器,能够帮我们自动分配和释放资源。而在异步方法中,with as 前面加上 async 代表声明一个支持异步的上下文管理器。

对于一些返回协程对象的操作,前面需要加 await 来修饰。例如 response 调用 text 方法,查询 API 可以发现,其返回的是协程对象,那么前面就要加 await;而对于状态码来说,其返回值就是一个数值,因此前面不需要加 await。所以,这里可以按照实际情况做处理,参考官方文档说明,看看其对应的返回值是怎样的类型,然后决定加不加 await 就可以了。

最后,定义完爬取方法之后,实际上是 main 方法调用了 fetch 方法。要运行的话,必须启用事件循环,而事件循环需要使用 asyncio 库,然后调用 run_until_complete 方法来运行。

注

在 Python 3.7及以后的版本中,我们可以使用 asyncio.run(main())代替最后的启动操作,不需要显示声明事件循环,run 方法内部会自动启动一个事件循环。但这里为了兼容更多的 Python 版本,依然显式声明了事件循环。

3. URL 参数设置

对于 URL 参数的设置,我们可以借助 params参数,传入一个字典即可,实例如下:

import aiohttp
import asyncio

async def main():
    params = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.get('https://www.httpbin.org/get', params=params) as response:
            print(await response.text())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

运行结果如下:

{
    "args": {
        "age": "25",
        "name": "germey"
    },
    "headers": {
        "Accept": "*/*",
        "Accept-Encoding": "gzip, deflate",
        "Host": "www.httpbin.org",
        "User-Agent": "Python/3.7 aiohttp/3.6.2",
        "X-Amzn-Trace-Id": "Root=1-5e85eed2-d240ac90f4dddf40b4723ef0"
    },
    "origin": "17.20.255.122",
    "url": "https://www.httpbin.org/get?name=germey&age=25"
}

这里可以看到,实际请求的URL 为 https://www.httpbin.org/get?name=germey&age=25,其中的参数对应于 params的内容。

4. 其他请求类型

aiohttp 还支持其他请求类型,如 POST、PUT、DELETE等,这些和 requests 的使用方式有点类似,实例如下:

session.post('http://www.httpbin.org/post', data=b'data')
session.put('http://www.httpbin.org/put', data=b'data')
session.delete('http://www.httpbin.org/delete')
session.head('http://www.httpbin.org/get')
session.options('http://www.httpbin.org/get')
session.patch('http://www.httpbin.org/patch', data=b'data')

要使用这些方法,只需要把对应的方法和参数替换一下。

5. POST 请求

对于POST 表单提交,其对应的请求头中的Content-Type 为application/x-www-form-urlencoded, 我们可以用如下方式来实现:

import aiohttp
import asyncio

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://www.httpbin.org/post', data=data) as response:
            print(await response.text())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

运行结果如下:

{
    "args": {},
    "data": "",
    "files": {},
    "form": {
        "age": "25",
        "name": "germey"
    },
    "headers": {
        "Accept": "*/*",
        "Accept-Encoding": "gzip, deflate",
        "Content-Length": "18",
        "Content-Type": "application/x-www-form-urlencoded",
        "Host": "www.httpbin.org",
        "User-Agent": "Python/3.7 aiohttp/3.6.2",
        "X-Amzn-Trace-Id": "Root=1-5e85f0b2-9017ea603a68dc285e0552do"
    },
    "json": null,
    "origin": "17.20.255.58",
    "url": "https://www.httpbin.org/post"
}

对于 POST JSON 数据提交,其对应的请求头中的Content-Type 为 application/json,我们只需要将 post 方法里的 data 参数改成json 即可,实例代码如下:

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://www.httpbin.org/post', json=data) as response:
            print(await response.text())

运行结果如下:

{
    "args": {},
    "data": "{\"name\": \"germey\", \"age\": 25}",
    "files": {},
    "form": {},
    "headers": {
        "Accept": "*/*",
        "Accept-Encoding": "gzip, deflate",
        "Content-Length": "29",
        "Content-Type": "application/json",
        "Host": "www.httpbin.org",
        "User-Agent": "Python/3.7 aiohttp/3.6.2",
        "X-Amzn-Trace-Id": "Root=1-5e85f03e-c91c9a20c79b9780dbed7540"
    },
    "json": {
        "age": 25,
        "name": "germey"
    },
    "origin": "17.20.255.58",
    "url": "https://www.httpbin.org/post"
}

可以发现,其实现也和 requests 非常像,不同的参数支持不同类型的请求内容。

6. 响应

对于响应来说,我们可以用如下方法分别获取其中的状态码、响应头、响应体、响应体二进制内容、响应体 JSON 结果,实例代码如下:

import aiohttp
import asyncio

async def main():
    data = {'name': 'germey', 'age': 25}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://www.httpbin.org/post', data=data) as response:
            print('status:', response.status)
            print('headers:', response.headers)
            print('body:', await response.text())
            print('bytes:', await response.read())
            print('json', await response.json())

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

运行结果如下:

status: 200
headers: <CIMultiDictProxy('Date': 'Thu, 02 Apr 2020 14:13:05 GMT', 'Content-Type': 'application/json',
'Content-Length': '503', 'Connection': 'keep-alive', 'Server': 'gunicorn/19.9.0',
'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true')>
body: {
    "args": {},
    "data": "",
    "files": {},
    "form": {
        "age": "25",
        "name": "germey"
    },
    "headers": {
        "Accept": "*/*",
        "Accept-Encoding": "gzip, deflate",
        "Content-Length": "18",
        "Content-Type": "application/x-www-form-urlencoded",
        "Host": "www.httpbin.org",
        "User-Agent": "Python/3.7 aiohttp/3.6.2",
        "X-Amzn-Trace-Id": "Root=1-5e85f2f1-f55326ff5800b15886c8e029"
    },
    "json": null,
    "origin": "17.20.255.58",
    "url": "https://www.httpbin.org/post"
}
bytes: b'{\n "args": {}, \n "data": "", \n "files": {}, \n "form":{\n "age": "25", \n "name": "germey"\n },
\n "headers": {\n "Accept": "*/*", \n "Accept-Encoding": "gzip, deflate", \n "Content-Length": "18",
\n "Content-Type": "application/x-www-form-urlencoded", \n "Host": "www.httpbin.org", \n "User-Agent":
"Python/3.7 aiohttp/3.6.2", \n "X-Amzn-Trace-Id": "Root=1-5e85f2f1-f55326ff5800b15886c8e029"\n }, \n
"json": null, \n "origin": "17.20.255.58", \n "url": "https://www.httpbin.org/post"\n}\n'
json: {'args': {}, 'data': '', 'files': {}, 'form': {'age': '25', 'name': 'germey'}, 'headers': {'Accept':
'*/*', 'Accept-Encoding': 'gzip, deflate', 'Content-Length': '18', 'Content-Type':
'application/x-www-form-urlencoded', 'Host': 'www.httpbin.org', 'User-Agent': 'Python/3.7 aiohttp/3.6.2',
'X-Amzn-Trace-Id': 'Root=1-5e85f2f1-f55326ff5800b15886c8e029'}, 'json': None, 'origin': '17.20.255.58',
'url': 'https://www.httpbin.org/post'}

可以看到,这里有些字段前面需要加 await,有些则不需要。其原则是,如果返回的是一个协程对象(如 async 修饰的方法),那么前面就要加 await,具体可以看 aiohttp 的 API,其链接为:https://docs. aiohttp.org/en/stable/client_reference.html。

7. 超时设置

我们可以借助 ClientTimeout 对象设置超时,例如要设置 1 秒的超时时间,可以这么实现:

import aiohttp
import asyncio
async def main():
    timeout = aiohttp.ClientTimeout(total=1)
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get('https://www.httpbin.org/get') as response:
            print('status:', response.status)

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

如果在1 秒之内成功获取响应,那么运行结果如下:

如果超时,则会抛出 TimeoutError异常,其类型为asyncio.TimeoutError,我们进行异常捕获即可。另外,声明 ClientTimeout 对象时还有其他参数,如 connect、socket_connect 等,详细可以参考官方文档:https://docs.aiohttp.org/en/stable/client_quickstart.html#timeouts。

8. 并发限制

由于aiohttp可以支持非常高的并发量,如几万、十万、百万都是能做到的,但面对如此高的并发量,目标网站很可能无法在短时间内响应,而且有瞬间将目标网站爬挂掉的危险,这提示我们需要控制一下爬取的并发量。一般情况下,可以借助 asyncio 的Semaphore 来控制并发量,实例代码如下:

import asyncio
import aiohttp

CONCURRENCY = 5
URL = 'https://www.baidu.com'

semaphore = asyncio.Semaphore(CONCURRENCY)
session = None

async def scrape_api():
    async with semaphore:
        print('scraping', URL)
        async with session.get(URL) as response:
            await asyncio.sleep(1)
            return await response.text()

async def main():
    global session
    session = aiohttp.ClientSession()
    scrape_index_tasks = [asyncio.ensure_future(scrape_api()) for _ in range(10000)]
    await asyncio.gather(*scrape_index_tasks)

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

这里我们声明 CONCURRENCY (代表爬取的最大并发量)为5,同时声明爬取的目标 URL 为百度。接着,借助 Semaphore 创建了一个信号量对象,将其赋值为semaphore,这样就可以用它来控制最大并发量了。怎么使用呢?这里我们把 semaphore 直接放置在了对应的爬取方法里,使用 async with 语句将 semaphore 作为上下文对象即可。这样一来,信号量便可以控制进入爬取的最大协程数量,即我们声明的 CONCURRENCY 的值。在 main 方法里,我们声明了10000 个 task,将其传递给gather 方法运行。倘若不加以限制,那这10000 个 task 会被同时执行,并发数量相当大。但有了信号量的控制之后,同时运行的 task 数量最大会被控制在5 个,这样就能给 aiohttp 限制速度了。

9. 总结

本节我们了解了 aiohttp 的基本使用方法,更详细的内容还是推荐大家查阅官方文档,详见

https://docs.aiohttp.org/。

本节代码参见:https://github.com/Python3WebSpider/AsyncTest。

6.3 aiohttp 异步爬取实战

6.2 节我们介绍了aiohttp 的基本用法,本节我们完成异步爬虫的实战演练。

1. 案例介绍

本次我们要爬取一个数据量相对大一点的网站,链接为https://spa5.scrape.center/, 页面如图6-1所示。

这是一个图书网站,整个网站包含数千本图书信息,网站数据是JavaScript 渲染而得的,数据可以通过 Ajax 接口获取,并且接口没有设置任何反爬措施和加密参数。另外,由于这个网站之前的电影案例数据量多一些,所以更加适合做异步爬取。本节我们要完成如下目标:

使用 aiohttp 爬取全站的图书数据;

将数据通过异步的方式保存到MongoDB 中。

2. 准备工作

开始本节的探索之前,请确保你已经做好了如下准备工作:

安装好了 Python(最低为Python3.6版本,最好为3.7版本或以上),并能成功运行Python程序;

了解了 Ajax 爬取的一些基本原理和模拟方法;

了解了异步爬虫的基本原理和 asyncio 库的基本用法;

了解了 aiohttp 库的基本用法;

安装并成功运行了 MongoDB 数据库,而且安装了异步爬虫库 motor。关于最后一条,要实现MongoDB 异步存储,离不开异步实现的MongoDB 存储库 motor,其安装命令为:

pip3 install motor

详细的安装方式可以参考:https://setup.scrape.center/motor。做好如上准备工作之后,我们就可以开始数据的爬取了。

3. 页面分析

第5章我们讲解了Ajax的基本分析方法,本节的案例站点和之前分析 Ajax时用的案例站点结构类似,都是列表页加详情页的结构,加载方式也都是Ajax,所以我们能轻松分析到如下信息。

列表页的 Ajax 请求接口格式为 https://spa5.scrape.center/api/book/?limit=18&offset={offset}。其中 limit 的值为每一页包含多少本书;offset 的值为每一页的偏移量,计算公式为 offset = limit * (page - 1),如第1页的offset 值为0,第2页offset的值为18,以此类推。

在列表页 Ajax 接口返回的数据里,results 字段包含当前页里18 本图书的信息,其中每本书的数据里都含有一个id 字段,这个id 就是图书本身的ID,可以用来进一步请求详情页。

详情页的 Ajax 请求接口格式为https://spa5.scrape.center/api/book/{id}。其中的id 即为详情页对应图书的ID,可以从列表页 Ajax 接口的返回结果中获取此内容。如果你掌握了5.3节的内容,那么上面三点应该很容易分析出来。如果有难度,不妨先复习一下之前的知识。

4. 实现思路

其实,一个完善的异步爬虫应该充分利用资源进行全速爬取,其实现思路是维护一个动态变化的爬取队列,每产生一个新的task,就将其放入爬取队列中,有专门的爬虫消费者从此队列中获取 task 并执行,能做到在最大并发量的前提下充分利用等待时间进行额外的爬取处理。但上面的实现思路整体较为烦琐,需要设计爬取队列、回调函数、消费者等机制,需要实现的功能较多。由于我们刚刚接触 aiohttp 的基本用法,本节也主要是了解aiohttp 的实战应用,因此这里稍微将爬取案例网站的实现过程简化一下。我们将爬取逻辑拆分成两部分,第一部分为爬取列表页,第二部分为爬取详情页。因为异步爬虫的关键点在于并发执行,所以可以将爬取拆分为如下两个阶段。

第一阶段是异步爬取所有列表页,我们可以将所有列表页的爬取任务集合在一起,并将其声明为由 task 组成的列表,进行异步爬取。

第二阶段则是拿到上一步列表页的所有内容并解析,将所有图书的id 信息组合为所有详情页的爬取任务集合,并将其声明为task 组成的列表,进行异步爬取,同时爬取结果也以异步方式存储到 MongoDB 里面。因为两个阶段在拆分之后需要串行执行,所以可能无法达到协程的最佳调度方式和资源利用情况,但也差不了很多。这个实现思路比较简单清晰,代码实现起来也较为容易,能够帮我们快速了解 aiohttp 的基本用法。

5. 基本配置

首先,先配置一些基本的变量并引入一些必需的库,代码如下:

import asyncio
import aiohttp
import logging

logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s: %(message)s')

INDEX_URL = 'https://spa5.scrape.center/api/book/?limit=18&offset={offset}'
DETAIL_URL = 'https://spa5.scrape.center/api/book/{id}'
PAGE_SIZE = 18
PAGE_NUMBER = 100
CONCURRENCY = 5

这里我们导入了 asyncio、aiohttp、logging这3 个库,然后定义了logging 的基本配置。接着定义了URL、爬取页码数量 PAGE_NUMBER、并发量 CONCURRENCY 等信息。

6. 爬取列表页

　　第一阶段来爬取列表页,还是和之前一样,先定义一个通用的爬取方法,代码如下:

semaphore = asyncio.Semaphore(CONCURRENCY)
session = None
async def scrape_api(url):
    async with semaphore:
        try:
            logging.info('scraping %s', url)
            async with session.get(url) as response:
                return await response.json()
        except aiohttp.ClientError:
            logging.error('error occurred while scraping %s', url, exc_info=True)

这里我们声明了一个信号量,用来控制最大并发数量。接着,定义了 scrape_api 方法,接收一个参数 url。该方法首先使用 async with 语句引入信号量作为上下文,接着调用 session 的get 方法请求这个url,然后返回响应的JSON 格式的结果。另外, 这里还进行了异常处理,捕获了 ClientError,如果出现错误,就会输出异常信息。

　　然后,爬取列表页,实现代码如下:

async def scrape_index(page):
    url = INDEX_URL.format(offset=PAGE_SIZE * (page - 1))
    return await scrape_api(url)

这里定义了 scrape_index 方法用于爬取列表页,它接收一个参数 page。随后构造了一个列表页的 URL,将其传给 scrape_api 方法即可。这里注意,方法同样需要用 async 修饰,调用的scrape_api方法前面需要加 await,因为scrape_api 调用之后本身会返回一个协程对象。另外,由于scrape_api的返回结果就是 JSON 格式,因此这个结果已经是我们想要爬取的信息,不需要再额外解析了。

　　接下来我们定义main 方法,将上面的方法串联起来调用,实现如下:

import json

async def main():
    global session
    session = aiohttp.ClientSession()
    scrape_index_tasks = [asyncio.ensure_future(scrape_index(page)) for page in range(1, PAGE_NUMBER + 1)]
    results = await asyncio.gather(*scrape_index_tasks)
    logging.info('results %s', json.dumps(results, ensure_ascii=False, indent=2))

if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

这里首先声明了 session 对象,即最初声明的全局变量。这样的话,就不需要在各个方法里面都传递 session了,实现起来比较简单。接着定义了 scrape_index_tasks,这就是用于爬取列表页的所有 task 组成的列表。然后调用 asyncio 的 gather 方法,并将 task 列表传入其参数,将结果赋值为 results,它是由所有 task 返回结果组成的列表。最后,调用 main 方法,使用事件循环启动该 main 方法对应的协程即可。

运行结果如下:

2020-04-03 03:45:54,692 - INFO: scraping https://spa5.scrape.center/api/book/?limit=18&offset=0
2020-04-03 03:45:54,707 - INFO: scraping https://spa5.scrape.center/api/book/?limit=18&offset=18
2020-04-03 03:45:54,707 - INFO: scraping https://spa5.scrape.center/api/book/?limit=18&offset=36
2020-04-03 03:45:54,708 - INFO: scraping https://spa5.scrape.center/api/book/?limit=18&offset=54
2020-04-03 03:45:54,708 - INFO: scraping https://spa5.scrape.center/api/book/?limit=18&offset=72
2020-04-03 03:45:56,431 - INFO: scraping https://spa5.scrape.center/api/book/?limit=18&offset=90
2020-04-03 03:45:56,435 - INFO: scraping https://spa5.scrape.center/api/book/?limit=18&offset=108

　　可以看到,这里就开始异步爬取了,并发量是由我们控制的,目前为5。当然,也可以进一步调高这个数字,在网站能承受的情况下,爬取速度会进一步加快。最后,results 就是爬取所有列表页得到的结果,接着就可以用它进行第二阶段的爬取了。

7. 爬取详情页

　　第二阶段是爬取详情页并保存数据。由于每个详情页分别对应一本书,每本书都需要一个ID 作为唯一标识,而这个ID 又正好存在 results 里面,所以下面我们需要将所有详情页的ID 获取出来。

　　在 main 方法里增加 results 的解析代码,实现如下:

ids = []
for index_data in results:
    if not index_data: continue
    for item in index_data.get('results'):
        ids.append(item.get('id'))

这样 ids 就是所有书的id 了,然后我们用所有的id 构造所有详情页对应的task,进行异步爬取即可。

　　这里再定义两个方法,用于爬取详情页和保存数据,实现如下:

from motor.motor_asyncio import AsyncIOMotorClient

MONGO_CONNECTION_STRING = 'mongodb://localhost:27017'
MONGO_DB_NAME = 'books'
MONGO_COLLECTION_NAME = 'books'

client = AsyncIOMotorClient(MONGO_CONNECTION_STRING)
db = client[MONGO_DB_NAME]
collection = db[MONGO_COLLECTION_NAME]

async def save_data(data):
    logging.info('saving data %s', data)
    if data:
        return await collection.update_one({
            'id': data.get('id')
        }, {
            '$set': data
        }, upsert=True)

async def scrape_detail(id):
    url = DETAIL_URL.format(id=id)
    data = await scrape_api(url)
    await save_data(data)

这里定义了 scrape_detail 方法用于爬取详情页数据,并调用save_data 方法保存数据。save_data 方法可以将数据保存到 MongoDB 里面。这里我们用到了支持异步的 MongoDB 存储库 motor。motor的连接声明和 pymongo 是类似的, 保存数据的调用方法也基本一致,不过整个都换成了异步方法。

　　接着在 main 方法里面增加对 scrape_detail 方法的调用即可爬取详情页,实现如下:

scrape_detail_tasks = [asyncio.ensure_future(scrape_detail(id)) for id in ids]
await asyncio.wait(scrape_detail_tasks)
await session.close()

这里先声明了 scrape_detail_tasks,这是由所有爬取详情页的 task 组成的列表,接着调用了 asyncio 的 wait 方法,并将声明的列表传入其中,调用执行此方法即可爬取详情页。当然,这里可以使用 gather 方法,效果是一样的,只不过返回结果略有差异。全部执行完毕后,调用 close 方法关闭 session。一些详情页的爬取过程如下:

2020-04-03 04:00:32,576 - INFO: scraping https://spa5.scrape.center/api/book/2301475
2020-04-03 04:00:32,576 - INFO: scraping https://spa5.scrape.center/api/book/2351866
2020-04-03 04:00:32,577 - INFO: scraping https://spa5.scrape.center/api/book/2828384
2020-04-03 04:00:32,577 - INFO: scraping https://spa5.scrape.center/api/book/3040352
2020-04-03 04:00:32,578 - INFO: scraping https://spa5.scrape.center/api/book/3074810
2020-04-03 04:00:44,858 - INFO: saving data {'id': '3040352', 'comments': [{'id': '387952888', 'content':
'温馨文,青梅竹马神马的很有爱~'}, ..., {'id': '2005314253', 'content':'沈晋&秦央,文比较短,平平淡淡,
贴近生活,短文的缺点不细腻'}], 'name':'那些风花雪月','authors': ['\n 公子欢喜'],'translators': [],
'tags': ['公子欢喜','耽美','BL','小说','现代','校园','那些风花雪月'],'url':'https://book.douban.com/subject/3040352/', 'isbn': '9789866685156', 'cover': 'https://img9.doubanio.com/view/subject/l/public/
s3029724.jpg', 'page_number': None, 'price': None, 'score': '8.1', 'introduction': '', 'catalog': None,
'published_at': '2008-03-26T16:00:00Z', 'updated_at': '2020-03-21T16:59:39.584722Z'}
2020-04-03 04:00:44,859 - INFO: scraping https://spa5.scrape.center/api/book/2994915
...

最后,我们观察到,爬取的数据都保存到 MongoDB 数据库里面了,如图6-2所示。

至此,我们就使用 aiohttp 完成了对图书网站的异步爬取。

8. 总结

本节我们通过一个实例讲解了 aiohttp 异步爬虫的具体实现。在学习过程中不难发现,相比普通的单线程爬虫来说,使用异步爬虫可以大大提高爬取效率,后面我们也会多多使用。本节代码参见:https://github.com/Python3WebSpider/ScrapeSpa5。

第6章 异步爬虫

6.1 协程的基本原理

1. 案例引入

2. 基础知识

阻塞

非阻塞

同步

异步

多进程

协程

3. 协程的用法

4. 准备工作

5. 定义协程

6. 绑定回调

7. 多任务协程

8. 协程实现

9. 使用 aiohttp

10. 总结

6.2 aiohttp 的使用

1. 基本介绍

2. 基本实例

3. URL 参数设置

4. 其他请求类型

5. POST 请求

6. 响应

7. 超时设置

8. 并发限制

9. 总结

6.3 aiohttp 异步爬取实战

1. 案例介绍

2. 准备工作

3. 页面分析

4. 实现思路

5. 基本配置

6. 爬取列表页

7. 爬取详情页

8. 总结