dede 分类信息网站 模板,vue做单页面网站,做网站不好做,法律问题咨询哪个网站做的好我们知道爬虫是 IO 密集型任务#xff0c;例如使用 requests 库来爬取某个站点#xff0c;当发出一个请求后#xff0c;程序必须等待网站返回响应#xff0c;才能接着运行#xff0c;而在等待响应的过程中#xff0c;整个爬虫程序是一直在等待的#xff0c;实际上没有做…我们知道爬虫是 IO 密集型任务例如使用 requests 库来爬取某个站点当发出一个请求后程序必须等待网站返回响应才能接着运行而在等待响应的过程中整个爬虫程序是一直在等待的实际上没有做任何事情。对于这种情况我们有没有优化方案呢?
当然有本篇博客我们就来了解一下异步爬虫的基本概念和实现。
要实现异步机制的爬虫那自然和协程脱不了关系。
案例引入
在介绍协程之前先来看一个案例网站地址为 https://www.httpbin.org/delay/5访问这个链接需要先等待五秒才能得到结果这是因为服务器强制等待了5秒时间才返回响应。
平时我们浏览网页的时候绝大部分网页的响应速度还是很快的如果写爬虫来爬取那么从发出请求到接收响应的时间不会很长因此需要我们等待的时间并不多。
然而像上面这个网站发出一次请求至少需要5秒才能得到响应如果用requests 库写爬虫来爬取那么每次都要等待5秒及以上才能拿到结果。
下面来测试一下我们用requests 写一个遍历程序直接遍历100 次案例网站试试看有什么效果实现代码如下:
import requests
import logging
import timelogging.basicConfig(levellogging.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个页面消耗的总时间。
运行结果如下:
2024-12-29 14:16:19,061-INFO:scraping https://www.httpbin.org/delay/5
2024-12-29 14:16:25,566-INFO:scraping https://www.httpbin.org/delay/5
2024-12-29 14:16:31,690-INFO:scraping https://www.httpbin.org/delay/5
2024-12-29 14:16:37,881-INFO:scraping https://www.httpbin.org/delay/5
2024-12-29 14:16:44,076-INFO:scraping https://www.httpbin.org/delay/5
...
2024-12-29 14:26:59,819-INFO:scraping https://www.httpbin.org/delay/5
2024-12-29 14:27:06,561-INFO:scraping https://www.httpbin.org/delay/5
2024-12-29 14:27:13,264-INFO:total time 654.2034454345703 seconds由于每个页面都至少要等待5秒才能加载出来因此100个页面至少要花费500秒时间加上网站本身的负载问题总的爬取时间最终约为654 秒大约11分钟。
这在实际情况中是很常见的有些网站本身加载速度就比较慢稍慢的可能1~3秒更慢的说不定10秒以上。如果我们就用requests单线程这么爬取总耗时将会非常大。此时要是打开多线程或多进程来爬取其爬取速度确实会成倍提升那么是否有更好的解决方案呢?
本节就来了解一下使用协程实现加速的方法这种方法对IO密集型任务非常有效。如过将其应用到网络爬虫中那么爬取效率甚至可以提升成百倍。
基础知识
了解协程需要先了解一些基础概念如阻塞和非阻塞、同步和异步、多进程和协程。
阻塞
阻塞状态指程序未得到所需计算资源时被挂起的状态。程序在等待某个操作完成期间自身无法继续干别的事情则称该程序在该操作上是阻塞的。
常见的阻塞形式有:网络 IO 阻塞、磁盘 IO阻塞、用户输入阻塞等。阻塞是无处不在的包括在 CPU切换上下文时所有进程都无法真正干事情它们也会被阻塞。在多核 CPU 的情况下正在执行上下文切换操作的核不可被利用。
非阻塞
程序在等待某操作的过程中自身不被阻塞可以继续干别的事情则称该程序在该操作上是非阻塞的。
非阻塞并不是在任何程序级别、任何情况下都存在的。仅当程序封装的级别可以囊括独立的子程序单元时程序才可能存在非阻塞状态。
非阻塞因阻塞的存在而存在正因为阻塞导致程序运行的耗时增加与效率低下我们才要把它变成非阻塞的。
同步
不同程序单元为了共同完成某个任务在执行过程中需要靠某种通信方式保持协调一致此时这些程序单元是同步执行的。
例如在购物系统中更新商品库存时需要用“行锁”作为通信信号强制让不同的更新请求排队并按顺序执行这里的更新库存操作就是同步的。
简言之同步意味着有序。
异步
为了完成某个任务有时不同程序单元之间无须通信协调也能完成任务此时不相关的程序单元之间可以是异步的。
例如爬取下载网页。调度程序调用下载程序后即可调度其他任务无须与该下载任务保持通信以协调行为。不同网页的下载、保存等操作都是无关的也无须相互通知协调。这些异步操作的完成时刻并不确定。
简言之异步意味着无序。
多进程
多进程就是利用 CPU 的多核优势在同一时间并行执行多个任务可以大大提高执行效率。
协程
协程英文叫作 coroutine又称微线程、纤程是一种运行在用户态的轻量级线程。
协程拥有自己的寄存器上下文和栈。协程在调度切换时将寄存器上下文和栈保存到其他地方等切回来的时候再恢复先前保存的寄存器上下文和栈。因此协程能保留上一次调用时的状态即所有局部状态的一个特定组合每次过程重人就相当于进人上一次调用的状态。
协程本质上是个单进程相对于多进程来说它没有线程上下文切换的开销没有原子操作锁定及同步的开销编程模型也非常简单。
我们可以使用协程来实现异步操作例如在网络爬虫场景下我们发出一个请求之后需要等待一定时间才能得到响应但其实在这个等待过程中程序可以干许多其他事情等得到响应之后再切换回来继续处理这样可以充分利用CPU和其他资源这就是协程的优势。
协程的用法
接下来我们了解一下协程的实现。从Python3.4开始Python 中加入了协程的概念但这个版本的协程还是以生成器对象为基础。Python 3.5 中增加了 async、await使得协程的实现更为方便。
Python 中使用协程最常用的库莫过于 asyncio所以本节会以它为基础来介绍协程的用法。
首先需要了解下面几个概念。 event loop:事件循环相当于一个无限循环我们可以把一些函数注册到这个事件循环上,当满足发生条件的时候就调用对应的处理方法。 coroutine:中文翻译叫协程在Python 中常指代协程对象类型我们可以将协程对象注册到事件循环中它会被事件循环调用。我们可以使用async 关键字来定义一个方法这个方法在调用时不会立即被执行而是会返回一个协程对象。 task:任务这是对协程对象的进一步封装包含协程对象的各个状态。 future:代表将来执行或者没有执行的任务的结果实际上和task 没有本质区别。
另外我们还需要了解 async、await 关键字它们是从 Python3.5才开始出现的专门用于定义协程。其中前者用来定义一个协程后者用来挂起阻塞方法的执行。
准备工作
在本节开始之前请确保安装的 Python 版本为 3.5 及以上如果版本是 3.4 及以下则下方的案例是不能运行的。
定义协程
我们来定义一个协程体验一下它和普通进程在实现上的不同之处代码如下:
import asyncioasync 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 0x000001F01B840740
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 asyncioasync def execute(x):print(Number:, x)return xcoroutine 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 0x000001D0096E0740
After calling execute
Task: Task pending nameTask-1 coroexecute() running at D:\projects\scrapy-demo\test.py:4
Number: 1
Task: Task finished nameTask-1 coroexecute() done, defined at D:\projects\scrapy-demo\test.py:4 result1
After calling loop这里我们定义了 loop 对象之后,紧接着调用了它的create_task方法,将协程对象转化为 task 对象随后打印输出一下发现它处于 pending 状态。然后将 task对象添加到事件循环中执行并再次打印出 task 对象发现它的状态变成了 finished同时还可以看到其 result 变成了 1也就是我们定义的 execute 方法的返回结果。
定义 task 对象还有另外一种方式就是直接调用 asyncio 包的 ensure_future 方法返回结果也是 task对象这样的话我们就可以不借助 loop 对象。即使还没有声明 loop也可以提前定义好 task对象这种方式的写法如下:
import asyncioasync def execute(x):print(Number:, x)return xcoroutine execute(1)
print(Coroutine:, coroutine)
print(After calling execute)
loop asyncio.get_event_loop()
task asyncio.ensure_future(coroutine)
print(Task:, task)
loop.run_until_complete(task)
print(Task:, task)
print(After calling loop)运行结果如下:
Coroutine: coroutine object execute at 0x00000295CC910740
After calling execute
Task: Task pending nameTask-1 coroexecute() running at D:\projects\scrapy-demo\test.py:4
Number: 1
Task: Task finished nameTask-1 coroexecute() done, defined at D:\projects\scrapy-demo\test.py:4 result1
After calling loop可以发现运行效果都是一样的。
绑定回调
我们也可以为某个 task 对象绑定一个回调方法。来看下面这个例子:
import asyncio
import requestsasync def request():url https://www.baidu.comstatus requests.get(url)return statusdef 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 nameTask-1 cororequest() running at D:\projects\scrapy-demo\test.py:5 cb[callback() at D:\projects\scrapy-demo\test.py:11]
Status: Response [200]
Task: Task finished nameTask-1 cororequest() done, defined at D:\projects\scrapy-demo\test.py:5 resultResponse [200]实际上即使不使用回调方法在 task运行完毕之后也可以直接调用result 方法获取结果代码如下所示:
import asyncio
import requestsasync def request():url https://www.baidu.comstatus requests.get(url)return statuscoroutine 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 nameTask-1 cororequest() running at D:\projects\scrapy-demo\test.py:5
Task: Task finished nameTask-1 cororequest() done, defined at D:\projects\scrapy-demo\test.py:5 resultResponse [200]
Task Result: Response [200]多任务协程
在上面的例子中我们都只执行了一次请求如果想执行多次请求应该怎么办呢?可以定义一个 task 列表然后使用 asyncio 包中的 wait 方法执行。看下面的例子:
import asyncio
import requestsasync def request():url https://www.baidu.comstatus requests.get(url)return statustasks [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 nameTask-1 cororequest() running at D:\projects\scrapy-demo\test.py:5, Task pending nameTask-2 cororequest() running at D:\projects\scrapy-demo\test.py:5, Task pending nameTask-3 cororequest() running at D:\projects\scrapy-demo\test.py:5, Task pending nameTask-4 cororequest() running at D:\projects\scrapy-demo\test.py:5, Task pending nameTask-5 cororequest() running at D:\projects\scrapy-demo\test.py:5]
Task Result: Response [200]
Task Result: Response [200]
Task Result: Response [200]
Task Result: Response [200]
Task Result: Response [200]可以看到5个任务被顺次执行并得到了执行结果。
协程实现
前面说了好一通又是 async 关键字又是 coroutine又是task又是 callback 的似乎并没有从中看出协程的优势反而写法上更加奇怪和麻烦了?别急上述案例只是为后面的使用作铺垫。接下来我们正式看看协程在解决 IO 密集型任务方面到底有怎样的优势。
在前面的代码中我们用一个网络请求作为例子这本身就是一个耗时等待操作因为在请求网页之后需要等待页面响应并返回结果。耗时等待操作一般都是IO操作例如文件读取、网络请求等。协程在处理这种操作时是有很大优势的当遇到需要等待的情况时程序可以暂时挂起转而执行其他操作从而避免因一直等待一个程序而耗费过多的时间能够充分利用资源。
为了表现协程的优势我们还是以本节开头介绍的网站 https://www.httpbin.org/delay/5 为例因为该网站响应比较慢所以可以通过爬取时间让大家直观感受到爬取速度的提升。
为了让大家更好地理解协程的正确使用方法这里先来看看大家使用协程时常犯的错误后面再给出正确的例子作为对比。
首先还是拿之前的requests 库进行网页请求之后再重新使用上面的方法请求一遍:
import asyncio
import requests
import timestart time.time()async def request():url https://www.httpbin.org/delay/5print(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]
Cost time: 76.17514300346375可以发现这和正常的请求并没有什么区别各个任务依然是顺次执行的耗时66秒平均一个请求耗时 6.6秒说好的异步处理呢?
其实要实现异步处理先得有挂起操作当一个任务需要等待IO 结果的时候可以挂起当前任务转而执行其他任务这样才能充分利用好资源。上面的方法都是一本正经地串行执行下来连个挂起都没有怎么可能实现异步?莫不是想太多了。
要实现异步我们再了解一下 await 关键字的用法,它可以将耗时等待的操作挂起让出控制权。如果协程在执行的时候遇到 await事件循环就会将本协程挂起转而执行别的协程直到其他协程挂起或执行完毕。
所以我们可能会将代码中的request 方法改成如下这样:
async def request():url https://www.httpbin.org/delay/5print(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
...
Cost time: 70.10251641273499
Task exception was never retrieved
future: Task finished nameTask-1 cororequest() done, defined at D:\projects\scrapy-demo\test.py:8 exceptionTypeError(object Response cant be used in await expression)
Traceback (most recent call last):File D:\projects\scrapy-demo\test.py, line 11, in requestresponse await requests.get(url)
TypeError: object Response cant be used in await expression这次协程遇到 await 时确实挂起了也等待了但是最后却报出以上错误信息。这个错误的意思是requests 返回的 Response 对象不能和 await 一起使用,为什么呢?因为根据官方文档说明,await后面的对象必须是如下格式之一:
一个原生协程对象;一个由types.coroutine 修饰的生成器这个生成器可以返回协程对象;由一个包含 await 方法的对象返回的一个迭代器。
这里 regeusts 返回的 Response 对象以上三种格式都不符合因此报出了上面的错误。
有的读者可能已经发现既然 await 后面可以跟一个协程对象那么 async 把请求的方法改成协程对象不就可以了吗?于是就代码被改写成如下的样子:
import asyncio
import requests
import timestart time.time()async def get(url):return requests.get(url)async def request():url https://www.httpbin.org/delay/5print(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]
...
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: 77.01466417312622协程还不是异步执行的也就是说我们仅仅将涉及IO 操作的代码封装到 async 修饰的方法里是不可行的。只有使用支持异步操作的请求方式才可以实现真正的异步这里 aiohttp 就派上用场了。
使用 aiohttp
aiohttp 是一个支持异步请求的库它和 asyncio 配合使用可以使我们非常方便地实现异步请求操作。 我们使用 pip3 安装即可:
pip3 install aiohttpaiohtp 的官方文档链接为 https://aiohttp.readthedocs.io/它分为两部分一部分是 Client一部分 是 Server。
下面我们将 aiohttp 投入使用将代码改写成如下样子:
import asyncio
import aiohttp
import requests
import timestart time.time()async def get(url):session aiohttp.ClientSession()response await session.get(url)await response.text()await session.close()return responseasync def request():url https://www.httpbin.org/delay/5print(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
...
Get response from https://www.httpbin.org/delay/5 response ClientResponse(https://www.httpbin.org/delay/5) [200 OK]
CIMultiDictProxy(Date: Sun, 29 Dec 2024 15:34:39 GMT, Content-Type: application/json, Content-Length: 367, Connection: keep-alive, Server: gunicorn/19.9.0, Access-Control-Allow-Origin: *, Access-Control-Allow-Credentials: true)Cost time: 8.169023752212524成功了!我们发现这次请求的耗时直接由76秒变成了8秒耗费时间减少了非常多。这里我们使用了 await其后面跟着 get 方法。在执行 10 个协程的时候如果遇到 await就会将当前协程挂起转而执行其他协程直到其他协程也挂起或执行完毕再执行下一个协程。
开始运行时事件循环会运行第一个 task。对于第一个 task 来说当执行到第一个 await 跟着的 get 方法时它会被挂起但这个get方法第一步的执行是非阻塞的挂起之后会立马被唤醒。立即又进人执行并创建了 clientSession对象。接着遇到第二个 await调用 session.get 请求方法然后就被挂起了。由于请求需要耗时很久所以一直没有被唤醒好在第一个 task 被挂起了那么接下来该怎么办呢?事件循环会寻找当前未被挂起的协程继续执行于是转而去执行第二个task流程操作和第一个 task 也是一样的以此类推直到执行第十个 task 的 session.get 方法之后全部的 task 都被挂起了。所有 task 都已经处于挂起状态那怎么办?只好等待了。5 秒之后几个请求几乎同时有了响应然后几个 task 也被唤醒接着执行并输出请求结果最后总耗时是8秒!
怎么样?这就是异步操作的便捷之处当遇到阻塞式操作时task被挂起程序接着去执行其他task而不是傻傻地等着这样可以充分利用 CPU而不必把时间浪费在等待IO 上。
有人会说在上面的例子中发出网络请求后接下来的5秒都是在等待这5秒之内CPU可以处理的 task数量远不止这些,既然这样的话,那么我们放 10个、20个、50个、100个、1000个 task一起执行最后得到所有结果的耗时不都是差不多的吗?因为这些任务被挂起后都是一起等待的。
从理论上来说确实是这样不过有个前提就是服务器即使在同一时刻接收无限次请求依然要能保证正常返回结果也就是服务器应该无限抗压另外还要忽略 0 传输时延。满足了这两点确实可以做到无限个 task 一起执行并且在预想时间内得到结果。但由于不同服务器处理 task 的实现机制不同可能某些服务器并不能承受那么高的并发量因此响应速度也会减慢。
这里我们以百度为例测试一下并发量分别为1、3、5、10、…、500 时的耗时情况代码如下
import asyncio
import timeimport aiohttpdef test(number):start time.time()async def get(url):session aiohttp.ClientSession()response await session.get(url)await response.text()await session.close()return responseasync def request():url https://www.baidu.comawait 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]:test(number)运行结果如下:
Number: 1 Cost time: 0.16419363021850586
Number: 3 Cost time: 0.13008475303649902
Number: 5 Cost time: 0.13481712341308594
Number: 10 Cost time: 0.14773201942443848
Number: 15 Cost time: 0.1420140266418457
Number: 30 Cost time: 0.15010547637939453
Number: 50 Cost time: 0.17553138732910156可以看到在服务器能够承受高并发的前提下即使我们增加了并发量其爬取速度也几乎不会太受影响。
综上所述使用了异步请求之后我们几乎可以在相同时间内实现成百上千倍次的网络请求把这个运用在爬虫中速度提升可谓非常可观。
