0x00:異步爬蟲概述

目的：在爬蟲中使用異步實現高性能的數據爬取操作。

先來看一個單線程、串行方式的爬蟲：

import requests headers = {'User-Agent':'xxx' } urls = {'xxxx''xxxx''xxxx' }def get_content(url):print("正在爬取：",url)#get方法是一個阻塞的方法reponse = requests.get(url=url,headers=headers)if reponse.status_code == 200:return reponse.content def parse_content(content):print("響應數據的長度為："，len(content))for url in urls:content = get_content(url)parse_content(content)

這段爬蟲就是一個一個去請求，如果前面在請求了后面的url只能等待，這樣爬取數據的效率會很低，所以使用異步操作爬取數據來提高效率。

0x01:多進程、多線程

優點：可以為相關阻塞的操作單獨開啟線程或進程，這樣阻塞操作就可以異步執行。
缺點：無法無限制的開啟多線程或多進程。

0x02:線程池、進程池

優點：可以降低系統對進程或者線程創建和銷毀的一個頻率，從而降低系統的開銷。
缺點：池中線程或進程的數量是有上限。

0x03:線程池的基本使用

先通過兩段爬蟲代碼進行對比一下

單線程串行方式執行

import timedef get_page(str):print("正在下載：",str)time.sleep(2)print("下載成功：",str)name_list = ['lemon','shy','good','nice']start_time = time.time()for i in range(len(name_list)):get_page(name_list[i])end_time = time.time() print('%d secode'% (end_time-start_time))

使用線程池的方式執行

import time #導入線程池模塊對應的類 from multiprocessing.dummy import Poolstart_time = time.time() def get_page(str):print("正在下載：",str)time.sleep(2)print("下載成功：",str)name_list = ['lemon','shy','good','nice']#實例化一個線程池對象 pool = Pool(4)#四個線程 #將列表中每一個列表元素傳遞給get_page進行處理 pool.map(get_page,name_list) end_time = time.time() print(end_time-start_time)

對比便可以看出線程池提高的效率

0x04:線程池應用

下面就通過爬取一些短視頻來進行練習

爬取生活區的這幾個視頻，先進行觀察，點開視頻都有詳細的url鏈接，所以先要爬取出各個視頻對應的詳情頁，再進行處理

ul標簽下的各個li標簽包含有a標簽，a標簽中的屬性又含有鏈接和名字，所以先將所有li標簽提取出來，再對各個鏈接和名稱進行處理，如果不想分析的話，也可以這樣

再稍微處理一下就可以了

import requests from lxml import etree if __name__ == '__main__':url = 'https://www.pearvideo.com/category_5'headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36'}page_text = requests.get(url=url,headers=headers).texttree = etree.HTML(page_text)li_list = tree.xpath('//*[@id="listvideoListUl"]/li')for li in li_list:detail_url = 'https://www.pearvideo.com/'+li.xpath('./div/a/@href')[0]name = li.xpath('./div/a/div[2]/text()')[0]+'.mp4'print(detail_url,name)

通過這段代碼，便可以得到視頻詳情頁和名稱，再來觀察視頻詳情頁

找到了鏈接所在的位置，但是當腳本爬取時發現返回空值，那查看一下視頻是不是動態加載出來的

剛才看到視頻的鏈接地址是在video標簽中的并且鏈接是以MP4結尾的，那就在響應數據中查找一下MP4


并不在video標簽中，而是一組JS代碼，因此這個video標簽一定是動態加載出來的，那接下來就去解析這個JS代碼，從JS代碼中提取出視頻鏈接，但是Xpath\bs4是無法處理JS數據的，所以使用正則表達式去解析

#將這串JS代碼復制下來 var contId="1677300",liveStatusUrl="liveStatus.jsp",liveSta="", playSta="1",autoPlay=!1,isLiving=!1,isVrVideo=!1,hdflvUrl="", sdflvUrl="",hdUrl="",sdUrl="",ldUrl="", srcUrl="https://video.pearvideo.com/mp4/third/20200529/cont-1677300-15195380-142357-hd.mp4", vdoUrl=srcUrl,skinRes="//www.pearvideo.com/domain/skin",videoCDN="//video.pearvideo.com"; #解析 ex = 'srcUrl="(.*?)",vdoUrl'

正則解析完成后，就可以得出這個視頻鏈接和名稱，那根據上面所了解的線程池的作用，處理阻塞且耗時的操作，請求視頻資源的話有的很大，所以會用到線程池如果進行持久化保存就更需要線程池，那就加上線程池，最終代碼如下：

import requests from lxml import etree import re from multiprocessing.dummy import Pool #需求:爬取梨視頻的視頻數據 #原則：線程池處理的是阻塞且耗時的操作 if __name__ == '__main__':url = 'https://www.pearvideo.com/category_5'headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36'}page_text = requests.get(url=url,headers=headers).texttree = etree.HTML(page_text)li_list = tree.xpath('//*[@id="listvideoListUl"]/li')urls = []#存儲所有視頻的鏈接和名稱for li in li_list:detail_url = 'https://www.pearvideo.com/'+li.xpath('./div/a/@href')[0]name = li.xpath('./div/a/div[2]/text()')[0]+'.mp4'# print(detail_url,name)#對詳情頁進行請求detail_page_text = requests.get(url=detail_url,headers=headers).text#解析出視頻的地址ex = 'srcUrl="(.*?)",vdoUrl'video_url = re.findall(ex,detail_page_text)[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.get(url=url,headers=headers).content#持久化存儲with open(dic['name'],'wb') as fp:fp.write(data)print(dic['name'],'下載成功！') #使用線程池對視頻數據進行請求（較為耗時的阻塞操作） pool = Pool(4) pool.map(get_video_data,urls) #關閉線程池 pool.close() #主線程等待子線程 pool.join()

0x05:單線程+異步協程

event_loop : 事件循環，相當于一個無限循環，可以把一些函數注冊到這個事件循環上，當滿足某些條件的時候，函數就會被循環執行。

coroutine：協程對象，可以將協程對象注冊到事件循環中，它會被事件循環調用。可以使用async關鍵字來定義一個方法，這個方法在調用時不會立即被執行，而是返回一個協程對象。

• task: 任務，它是協程對象的進一步封裝，包含了任務的各個狀態。
• future: 代表將來執行或還沒有執行的任務，實際上和task沒有本質區別。
• async：定義一個協程。
• await: 用來掛起阻塞方法的執行。

簡單的一個例子：

import asyncioasync def result(url):print("正在請求的url是",url)print("請求成功",url) #async修飾的函數，調用之后返回一個協程對象 c = result('www.baidu.com')#創建一個事件循環對象 loop = asyncio.get_event_loop()#需要將協程對象注冊到循環事件中，啟動事件循環 #run_until_complete函數即可以啟動又可以注冊 loop.run_until_complete(c)

再看一下task的使用

import asyncioasync def result(url):print("正在請求的url是",url)print("請求成功",url) #async修飾的函數，調用之后返回一個協程對象 c = result('www.baidu.com')#task的使用 loop = asyncio.get_event_loop() #基于loop創建一個task對象 task = loop.create_task(c) print(task) #注冊到事件循環當中 loop.run_until_complete(task) print(task)

future的使用

import asyncioasync def result(url):print("正在請求的url是",url)print("請求成功",url) #async修飾的函數，調用之后返回一個協程對象 c = result('www.baidu.com')#future loop = asyncio.get_event_loop() future = asyncio.ensure_future(c) print(future) loop.run_until_complete(future) print(future)

回調函數的使用

創建回調函數 add_done_callback import asyncioasync def result(url):print("正在請求的url是",url)print("請求成功",url)return url #async修飾的函數，調用之后返回一個協程對象 c = result('www.baidu.com') #創建一個回調函數 def callback_func(task):#result返回的就是任務對象中封裝的協程對象對應函數的返回值print(task.result()) #綁定回調 loop = asyncio.get_event_loop() task = asyncio.ensure_future(c) #將回調函數綁定到任務對象中 #默認將task對象傳遞給callback_func函數 task.add_done_callback(callback_func)loop.run_until_complete(task)

0x06:多任務異步協程實現

import asyncio import timeasync def request(url):print("正在下載",url)#在異步協程中如果出現了同步模塊相關的代碼，就無法實現異步操作# time.sleep(2)#當asyncio遇到阻塞操作時，必須進行手動掛起await asyncio.sleep(2)print("下載完畢",url)start = time.time() urls = ['www.baidu.com','www.sogou.com','www.bing.com' ]#任務列表：存放多個任務對象 stasks = [] for url in urls:#協程對象c = request(url)#封裝到任務對象中task = asyncio.ensure_future(c)stasks.append(task) loop = asyncio.get_event_loop() #固定寫法，不能直接將task列表直接放入，封裝到wait中 loop.run_until_complete(asyncio.wait(stasks)) print(time.time()-start)

最終的下載時間

0x07:aiohttp模塊

先創建一個web服務，用于爬取測試

from flask import Flask import timeapp = Flask(__name__)@app.route('/a') def index_a():time.sleep(2)return 'Hello lemon'@app.route('/b') def index_b():time.sleep(2)return 'Hello shy'@app.route('/c') def index_c():time.sleep(2)return 'Hello theshy'if __name__ == '__main__':app.run(threaded=True)

編寫爬取腳本：

import asyncio import requests import timestart = time.time() urls = ['http://127.0.0.1:5000/a','http://127.0.0.1:5000/b','http://127.0.0.1:5000/c' ]async def get_page(url):print("正在爬取", url)reponse = requests.get(url)print("爬取成功",reponse.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_complete(asyncio.wait(tasks)) end = time.time()print('總耗時',end-start)

剛開始這樣寫覺得也沒什么問題，但一運行發現不是異步操作

問題就出現在requests.get這個地方 ,requests.get是基于同步，必須使用基于異步的網絡請求模塊進行請求url的請求發送,所以就需要使用aiohttp模塊

import asyncio import requests import time import aiohttpstart = time.time() urls = ['http://127.0.0.1:5000/a','http://127.0.0.1:5000/b','http://127.0.0.1:5000/c' ]async def get_page(url):async with aiohttp.ClientSession() as session:#headers,params/data,proxy='http://ip:port'#請求有阻塞，需要手動掛起async with await session.get(url) as reponse:#text()返回字符串形式的響應數據#read()返回二進制形式的響應數據#json()返回的就是json對象#注意：在獲取響應數據操作之前一定要使用await進行手動掛起page_text = await reponse.text()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_complete(asyncio.wait(tasks)) end = time.time()print('總耗時',end-start)

總結

以上是生活随笔為你收集整理的Python爬虫之旅_高性能异步爬虫的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。