异步爬虫详解

异步爬虫

目的实现高性能数据爬取操作

原则线程池处理的是较为阻塞且耗时的操作

异步爬虫的方式

多线程、多进程（不建议）

好处可以为相关阻塞的操作单独开启线程或进程，阻塞操作就可以异步执行。

弊端无法无限地开启多线程或多进程。

线程池、进程池（适当的使用）

好处可以降低系统对进程或线程创建与销毁的频率，从而很好地降低系统的开销。

弊端池中进程或线程地数量是又上限的。

实例

单线程串行方式

import time
def get_page(str):
    print("正在下载", str)
    time.sleep(2)
    print("下载成功", str)
​
name_list = ["aa", "bb", "", "dd"]
start_time = time.time()
for i in range(len(name_list)):
    get_page(name_list[i])
end_time = time.time()
print("%d second"%(end_time - start_time))
​
运行结果
正在下载 aa
下载成功 aa
正在下载 bb
下载成功 bb
正在下载 
下载成功 
正在下载 dd
下载成功 dd
8 second

线程池处理方式（注解是重点）

import time
from multiprocessing.dummy import Pool//导入线程池模块对应的类
start_time = time.time()
def get_page(str):
    print("正在下载", str)
    time.sleep(2)
    print("下载成功", str)
name_list = ["aa", "bb", "", "dd"]
pool = Pool(4)  //实例化一个线程池对象
pool.map(get_page, name_list)  //将列表中每一个列表元素传递给get_page函数进行处理
end_time = time.time()
print("%d second"%(end_time - start_time))
​
运行结果
正在下载 aa
正在下载 bb
正在下载 
正在下载 dd
下载成功 aa
下载成功 bb
下载成功 
下载成功 dd
2 second

实战演练

from lxml import etree
from multiprocessing.dummy import Pool
import re
import requests
headers = {
    "user-agent":"Mozilla/5.0 (Windos NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/95.0.4638.54 Safari/537.36 Edg/95.0.1020.40"
​
}
url = 'https://.pearvideo./category_5'
page_text = requests.post(url=url, headers=headers).text
tree = etree.HTML(page_text)
li_list = tree.xpath('//[@id="listvideoListUl"]/li')
urls = []
for li in li_list[0:3]:
    detail_li_url = 'https://.pearvideo./' + li.xpath('.//a/@href')[0]
    name = li.xpath('.//div[@]/text()')[0]+".jpg"
    #详情页的url发起请求
    detail_page_ajax = requests.get(url=detail_li_url, headers=headers).text
    #从详情页中解析出视频的地址(url)
    ex = 'img  src="http://.kao./skin/sinaskin/image/nopic.gif"'
    video_url = re.findall(ex, detail_page_ajax)[0]
    print(video_url)
​
    dic={
        'name': name,
        'url': video_url
    }
    urls.append(dic)
#对视频链接发起请求获取视频的二进制数据，然后将视频数据进行返回
def get_video_data(dic):
    url = dic['url']
    print(dic['name'], '正在下载......')
    data = requests.post(url=url, headers=headers).content
    #持久化存储
    ith open(dic['name'], 'b') as fp:
        fp.rite(data)
        print(dic['name'], '下载成功！')
​
pool = Pool(5)
pool.map(get_video_data, urls)
pool.close()
pool.join()
​
运行结果
https://image1.pearvideo./cont/20220113/cont-1749918-71026328.png
https://image1.pearvideo./cont/20220113/cont-1703855-12643918.png
https://image.pearvideo./cont/20220113/11643363-120751-1.png
90后沪漂每天横跨上海通勤4小时不需要逃离北上广.jpg 正在下载......
美国公司推出海上住宅服务足不出户也能环游世界.jpg  正在下载......
海淀全年全域禁止燃放烟花爆竹～向烟花爆竹说No，从你做起！.jpg 正在下载......
海淀全年全域禁止燃放烟花爆竹～向烟花爆竹说No，从你做起！.jpg 下载成功！
美国公司推出海上住宅服务足不出户也能环游世界.jpg 下载成功！
90后沪漂每天横跨上海通勤4小时不需要逃离北上广.jpg 下载成功！
​
Process finished ith exit code 0

单线程+异步协程（推荐）

event_loop事件循环，相当于一个无限循环，我们可以把一些函数注册到这个事件循环上。当满足某些条件的时候，函数就会被循环执行。

注释是精华

import asyncio
async def request(url):
    print("正在请求的url是", url)
    print("请求成功，", url)
​
c = request(".baidu.")
#async 修饰的函数，调用之后返回一个协程对象
​
loop = asyncio.get_event_loop()
#创建一个事件循环对象
​
loop.run_until_plete(c)
#将协程对象注册到loop中,然后启动loop
​
运行结果
正在请求的url是 .baidu.
请求成功， .baidu.

coroutine协程对象，我们可以将协程对象注册到事件循环中，他会被事件循环调用。我们可以使用async关键字来定义一个方法，这个方法在调用时不用立即被执行，而是返回一个协程对象。

task任务，它是对协程对象的进一步封装，包含了任务的各个状态。

task的使用（注释是精华）

import asyncio
async def request(url):
    print("正在请求的url是", url)
    print("请求成功，", url)
​
c = request(".baidu.")
#async 修饰的函数，调用之后返回一个协程对象
​
loop = asyncio.get_event_loop()
#创建一个事件循环对象
​
task = loop.create_task(c)
#基于loop对象创建一个task对象
​
print("当前task的状态是", task)
loop.run_until_plete(task)
print("当前task的状态是", task)
​
运行结果
当前task的状态是 >
正在请求的url是 .baidu.
请求成功， .baidu.
当前task的状态是  result=None>
​

future代表将来执行或还没有执行的任务，实际上和task没有本质区别。

future的使用（注释是精华）

import asyncio
async def request(url):
    print("正在请求的url是", url)
    print("请求成功，", url)
​
c = request(".baidu.")
#async 修饰的函数，调用之后返回一个协程对象
​
loop = asyncio.get_event_loop()
#创建一个事件循环对象
​
task = asyncio.ensure_future(c)
#通过asyncio创建future
​
print("当前task的状态是", task)
loop.run_until_plete(task)
print("当前task的状态是", task)

async定义一个协程。

aait用来挂起阻塞方法的执行。

绑定回调（注解是精华）

import asyncio
async def request(url):
    print("正在请求的url是", url)
    print("请求成功，", url)
    return "Hello World!"
​
c = request(".baidu.")
#async 修饰的函数，调用之后返回一个协程对象
​
def callback_func(task):
    #绑定回调函数
    print (task.result())
    #return 返回的是任务对象中封装的协程对象对应函数的返回值(关键！！！！)
​
loop = asyncio.get_event_loop()
#创建一个事件循环对象
​
task = asyncio.ensure_future(c)
​
task.add_done_callback(callback_func)
#将回调函数绑定到任务对象中
​
loop.run_until_plete(task)

多任务协程1.0（注释都是精华）

import asyncio
import time
async def request(url):
    print("正在请求的url是", url)
    #time.sleep(2)
    #在异步协程中，如果出现了同步模块相关的代码，那么就无法实现异步，如上，运行结果是:6s
​
    aait  asyncio.sleep(2)
    #当在asyncio中遇到阻塞操作必须进行手动挂起
    
    print("请求成功，", url)
​
start_time = time.time()
urls = [
    ".douban.",
    ".baidu.",
    ".sougu."
]
​
tasks = []
#任务列表存放多个任务对象
​
for url in urls:
    c = request(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)
​
loop = asyncio.get_event_loop()
​
loop.run_until_plete(asyncio.ait(tasks))
#需要将任务列表封装到ait中
​
print(time.time()-start_time)
​
运行结果
正在请求的url是 .douban.
正在请求的url是 .baidu.
正在请求的url是 .sougu.
请求成功， .douban.
请求成功， .baidu.
请求成功， .sougu.
2.0095303058624268
​
运行结果（#time.sleep(2)）
正在请求的url是 .douban.
请求成功， .douban.
正在请求的url是 .baidu.
请求成功， .baidu.
正在请求的url是 .sougu.
请求成功， .sougu.
6.0322794914245605

多任务协程2.0（注释都是精华）（自定义服务器）

自定义的flask服务器

from flask import Flask
import time

app = Flask(__name__)

@app.route('/bobo')
def index_bobo():
 time.sleep(2)
 return "Hello bobo"

@app.route('/jay')
def index_jay():
 time.sleep(2)
 return "Hello jay"

@app.route('/tom')
def index_tom():
 time.sleep(2)
 return "Hello tom"

if __name__ == '__main__':
 app.run(threaded=True)

多任务异步协程（失败案例）（注解都是精华）

  ```py
  import asyncio
  import requests
  import time
  
  start_time = time.time()
  urls = [
   'http://127.0.0.1:5000/bobo',
   'http://127.0.0.1:5000/jay',
   'http://127.0.0.1:5000/tom'
  ]
  async def get_page(url):
   print("正在下载", url)
   #request.get是基于同步，必须使用基于异步的网络请求模块进行指定url的请求发送
   #aiohttp:基于异步网络请求的模块
   response = requests.get(url=url)
   print("下载完毕", response.text)
  
  tasks = []
  for url in urls:
   c = get_page(url)
   task = asyncio.ensure_future(c)
   tasks.append(task)
  
  loop = asyncio.get_event_loop()
  loop.run_until_plete(asyncio.ait(tasks))
  
  end_time = time.time()
  print("总耗时", end_time-start_time)
  
  运行结果
  正在下载 http://127.0.0.1:5000/bobo
  下载完毕 Hello bobo
  正在下载 http://127.0.0.1:5000/jay
  下载完毕 Hello jay
  正在下载 http://127.0.0.1:5000/tom
  下载完毕 Hello tom
  总耗时 6.023055553436279
  ```

多任务异步协程（成功案例）（注解都是精华）

import asyncio
import requests
import time
import aiohttp

start_time = time.time()
urls = [
 'http://127.0.0.1:5000/bobo',
 'http://127.0.0.1:5000/jay',
 'http://127.0.0.1:5000/tom'
]
async def get_page(url):
 async ith  aiohttp.ClientSession() as session:
  async ith aait session.get(url) as response:
  # 这里的session.get/post()的用法跟request.get/post()的用法类似，
  # 唯一的区别是在使用代理时，要采用该形式proxy="http://ip:port
   page_text = aait response.text()
   #text()返回字符串形式的响应数据
   #read()返回二进制形式的响应数据
   #注意获取响应数据操作之前一定要使用aait进行手动挂起
   print(page_text)


tasks = []
for url in urls:
 c = get_page(url)
 task = asyncio.ensure_future(c)
 tasks.append(task)

loop = asyncio.get_event_loop()
loop.run_until_plete(asyncio.ait(tasks))

end_time = time.time()
print("总耗时", end_time-start_time)

运行结果
Hello bobo
Hello tom
Hello jay
总耗时 2.0107059478759766