異步爬蟲(chóng)：

• 基于線程池
• 基于單線程+多任務(wù)的異步爬蟲(chóng)

線程池

• from multiprocessing.dummy import Pool

• map（callback，alist）

  • 可以使用callback對(duì)alist中的每一個(gè)元素進(jìn)行指定形式的異步操作

• 為了體現(xiàn)效果，我們自己搭建一個(gè)web服務(wù)

  • Flask的基本使用
  • 環(huán)境安裝 pip install flask
  • 創(chuàng)建一個(gè)py源文件

test.html


自定義一個(gè)簡(jiǎn)單網(wǎng)頁(yè)

# -*- coding: utf-8 -*- from flask import Flask from flask import render_template import time#實(shí)例化一個(gè)app app = Flask(__name__)# 創(chuàng)建視圖函數(shù)&路由地址 @app.route("/gpc") #路由地址 def index_1(): # 視圖函數(shù)time.sleep(2)#為了表示堵塞效果我們定義一個(gè)相應(yīng)時(shí)間return render_template("test.html")@app.route("/una") def index_2():time.sleep(2)return render_template("test.html")@app.route("/python") def index_3():time.sleep(2)return render_template("test.html")if __name__=="__main__":#debug=True 表示開(kāi)啟調(diào)試模式：服務(wù)器端代碼被修改后按下保存鍵會(huì)自動(dòng)重啟服務(wù)app.run(debug=True)

同步爬蟲(chóng)

import requests import timedef get_request(url):page_text = requests.get(url=url).textreturn len(page_text)#同步代碼總耗時(shí): 6.035839080810547秒 if __name__ == "__main__":start = time.time()#開(kāi)始時(shí)間urls = ["http://127.0.0.1:5000/gpc","http://127.0.0.1:5000/una","http://127.0.0.1:5000/python"]for url in urls:res = get_request(url)print(res)print("總耗時(shí):",time.time()-start)

異步爬蟲(chóng)

import requests from multiprocessing.dummy import Pool #線程池 import timedef get_request(url):page_text = requests.get(url=url).textreturn len(page_text)urls = ["http://127.0.0.1:5000/gpc","http://127.0.0.1:5000/una","http://127.0.0.1:5000/python"] # 異步代碼 if __name__ == "__main__":start = time.time() # 開(kāi)始時(shí)間pool = Pool(3)#3表示開(kāi)啟的現(xiàn)場(chǎng)數(shù)量# get_request作為回調(diào)函數(shù)，需要基于異步的形式對(duì)urls列表中的每一個(gè)列表元素進(jìn)行操作# 保證回調(diào)函數(shù)必須要有一個(gè)參數(shù)和返回值result_list = pool.map(get_request,urls)print(result_list)print("總耗時(shí):", time.time() - start)# # [45532, 45532, 45532]# 總耗時(shí): 2.023587703704834

異步爬蟲(chóng)數(shù)據(jù)解析

import requests from multiprocessing.dummy import Pool#線程池 from lxml import etree import timedef get_request(url):page_text = requests.get(url=url).texttree = etree.HTML(page_text)analysis = tree.xpath("//div[@id='contson09d31b73b44d']/p[5]//text()")#定義xpath數(shù)據(jù)解析return analysisurls = ["http://127.0.0.1:5000/gpc","http://127.0.0.1:5000/una","http://127.0.0.1:5000/python"] # 異步代碼 if __name__ == "__main__":start = time.time() # 開(kāi)始時(shí)間pool = Pool(3)#3表示開(kāi)啟的現(xiàn)場(chǎng)數(shù)量# get_request作為回調(diào)函數(shù)，需要基于異步的形式對(duì)urls列表中的每一個(gè)列表元素進(jìn)行操作# 保證回調(diào)函數(shù)必須要有一個(gè)參數(shù)和返回值result_list = pool.map(get_request,urls)print(result_list)print("總耗時(shí):", time.time() - start)## [['臨別殷勤重寄詞，詞中有誓兩心知。', '七月七日長(zhǎng)生殿，夜半無(wú)人私語(yǔ)時(shí)。', '在天愿作比翼鳥(niǎo)，在地愿為連理枝。', '天長(zhǎng)地久有時(shí)盡，此恨綿綿無(wú)絕期。'],# ['臨別殷勤重寄詞，詞中有誓兩心知。', '七月七日長(zhǎng)生殿，夜半無(wú)人私語(yǔ)時(shí)。', '在天愿作比翼鳥(niǎo)，在地愿為連理枝。', '天長(zhǎng)地久有時(shí)盡，此恨綿綿無(wú)絕期。'],# ['臨別殷勤重寄詞，詞中有誓兩心知。', '七月七日長(zhǎng)生殿，夜半無(wú)人私語(yǔ)時(shí)。', '在天愿作比翼鳥(niǎo)，在地愿為連理枝。', '天長(zhǎng)地久有時(shí)盡，此恨綿綿無(wú)絕期。']]# 總耗時(shí): 2.0326013565063477

單線程+多任務(wù)的異步爬蟲(chóng)

• 單線程+多任務(wù)的異步爬蟲(chóng)
• pip install asyncio
  • 特殊的函數(shù)
    • 如果一個(gè)函數(shù)的定義被async修飾后，則該函數(shù)就變成了一個(gè)特殊函數(shù)
    • 該特殊函數(shù)調(diào)用后，函數(shù)內(nèi)部的實(shí)現(xiàn)語(yǔ)句不會(huì)被【立即】執(zhí)行
  • 協(xié)程對(duì)象
    • 對(duì)象：通過(guò)特殊函數(shù)的調(diào)用返回一個(gè)協(xié)程對(duì)象
    • 協(xié)程對(duì)象 == 特殊函數(shù) == 一組指定的操作
    • 協(xié)程 == 一組指定的操作
  • 任務(wù)對(duì)象
    • 任務(wù)對(duì)象就是一個(gè)高級(jí)的協(xié)程對(duì)象。（任務(wù)對(duì)象就是對(duì)協(xié)程的進(jìn)一步封裝）
    • 任務(wù)對(duì)象 == 協(xié)程對(duì)象 == 特殊函數(shù) == 一組指定的操作
    • 任務(wù)對(duì)象 == 一組指定的操作
    • 如何創(chuàng)建一個(gè)任務(wù)對(duì)象
      • asyncio.ensure_future(協(xié)程對(duì)象）
    • 任務(wù)對(duì)象的高級(jí)之處
      • 可以給任務(wù)對(duì)象綁定一個(gè)回調(diào)：
        • 給任務(wù)對(duì)象綁定回調(diào)：
          • task.add_done_callback(task_callback())
          • 回調(diào)函數(shù)的調(diào)用時(shí)機(jī)：
            • 在我們?nèi)蝿?wù)執(zhí)行結(jié)束后，才可以調(diào)用回調(diào)函數(shù)
          • 回調(diào)函數(shù)的參數(shù)只可以有一個(gè)：表示的就是該回調(diào)函數(shù)的調(diào)用者（任務(wù)對(duì)象）
          • 使用回調(diào)函數(shù)的參數(shù)調(diào)用result（）返回的就是任務(wù)對(duì)象表示的特殊函數(shù)return的結(jié)果
  • 事件循環(huán)對(duì)象
    • 對(duì)象
    • 作用
      • 可以將多個(gè)任務(wù)對(duì)象注冊(cè)/裝載到事件循環(huán)對(duì)象中
      • 如果開(kāi)啟了事件循環(huán)后，則其內(nèi)部注冊(cè)/裝載的任務(wù)對(duì)象表示的指定操作就會(huì)被基于異步被執(zhí)行
    • 事件循環(huán)對(duì)象創(chuàng)建
      • loop = asyncio.get_event_loop()
    • 注冊(cè)且啟動(dòng)
      • loop.run_until_complete(task)
  • wait 方法的作用
    • 將任務(wù)列表中的任務(wù)對(duì)象賦予可被掛起的權(quán)限，只有任務(wù)對(duì)象被賦予了可被掛起的權(quán)限后，該任務(wù)對(duì)象才可以被掛起
    • 掛起：表示將當(dāng)前的任務(wù)對(duì)象交出cpu的使用權(quán)
  • await 關(guān)鍵字
    • 在特殊函數(shù)內(nèi)部，凡是阻塞操作前必須使用await進(jìn)行修飾，await就可以保證阻塞操作在異步執(zhí)行的過(guò)程中不會(huì)被跳過(guò)！
• 注意事項(xiàng)【重要】
  • 在特殊函數(shù)內(nèi)部不可以出現(xiàn)不支持異步模塊對(duì)應(yīng)的代碼，否則會(huì)中斷整個(gè)異步效果

單任務(wù)操作：

import asyncio import timeasync def get_request(url):print("正在請(qǐng)求的url:",url)time.sleep(2)print("請(qǐng)求結(jié)束:",url) if __name__=="__main__":c = get_request("www.1.com") # c就是一個(gè)協(xié)程對(duì)象task =asyncio.ensure_future(c)#task就是一個(gè)任務(wù)對(duì)象，它是對(duì)協(xié)程對(duì)象的進(jìn)一步封裝loop = asyncio.get_event_loop() #loop就是一個(gè)事件循環(huán)對(duì)象loop.run_until_complete(task) # 將任務(wù)對(duì)象注冊(cè)到事件循環(huán)中且開(kāi)啟事件循環(huán)

綁定回調(diào)函數(shù)

import asyncio import timeasync def get_request(url):print("正在請(qǐng)求的url:",url)time.sleep(2)print("請(qǐng)求結(jié)束:",url)return "我是特殊函數(shù)的返回值"#回調(diào)函數(shù)的封裝 def task_callback(t):#參數(shù)t：就是該回調(diào)函數(shù)的調(diào)用者(任務(wù)對(duì)象)print("i am task_callback(),參數(shù)t:",t)print("t.result()返回的是:",t.result())if __name__=="__main__":c = get_request("www.1.com") # c就是一個(gè)協(xié)程對(duì)象task =asyncio.ensure_future(c)#task就是一個(gè)任務(wù)對(duì)象，它是對(duì)協(xié)程對(duì)象的進(jìn)一步封裝task.add_done_callback(task_callback) #給task綁定一個(gè)回調(diào)函數(shù)loop = asyncio.get_event_loop() #loop就是一個(gè)事件循環(huán)對(duì)象loop.run_until_complete(task) # 將任務(wù)對(duì)象注冊(cè)到事件循環(huán)中且開(kāi)啟事件循環(huán)

多任務(wù)操作

import asyncio import timeasync def get_request(url):print("正在請(qǐng)求的url:",url)time.sleep(2)#（堵塞操作2秒）出現(xiàn)了不支持異步模塊的對(duì)象所以耗時(shí)6秒未成功進(jìn)行異步操作print("請(qǐng)求結(jié)束:",url)urls = ["http://127.0.0.1:5000/gpc","http://127.0.0.1:5000/una","http://127.0.0.1:5000/python" ]if __name__=="__main__":start = time.time()tasks = [] # 多任務(wù)列表# 1.創(chuàng)建協(xié)程對(duì)象for url in urls:c = get_request(url)# 2.創(chuàng)建任務(wù)對(duì)象task = asyncio.ensure_future(c)tasks.append(task)# 3.創(chuàng)建事件循環(huán)對(duì)象loop = asyncio.get_event_loop()# 4.將任務(wù)對(duì)象注冊(cè)到事件循環(huán)中且開(kāi)啟事件循環(huán)# loop.run_until_complete(tasks)#循環(huán)中不能放列表loop.run_until_complete(asyncio.wait(tasks))# 必須使用wait方法對(duì)tasks封裝print("總耗時(shí):",time.time()-start)

多任務(wù)異步

import asyncio import timeasync def get_request(url):print("正在請(qǐng)求的url:",url)await asyncio.sleep(2)#（堵塞操作2秒）支持異步模塊的代碼print("請(qǐng)求結(jié)束:",url)urls = ["http://127.0.0.1:5000/gpc","http://127.0.0.1:5000/una","http://127.0.0.1:5000/python" ]if __name__=="__main__":start = time.time()tasks = [] # 多任務(wù)列表# 1.創(chuàng)建協(xié)程對(duì)象for url in urls:c = get_request(url)# 2.創(chuàng)建任務(wù)對(duì)象task = asyncio.ensure_future(c)tasks.append(task)# 3.創(chuàng)建事件循環(huán)對(duì)象loop = asyncio.get_event_loop()# 4.將任務(wù)對(duì)象注冊(cè)到事件循環(huán)中且開(kāi)啟事件循環(huán)# loop.run_until_complete(tasks)#循環(huán)中不能放列表loop.run_until_complete(asyncio.wait(tasks))# 必須使用wait方法對(duì)tasks封裝print("總耗時(shí):",time.time()-start)

協(xié)程爬蟲(chóng)實(shí)例

requests是一個(gè)不支持異步的模塊
aiohttp

• aiohttp是一個(gè)支持異步的網(wǎng)絡(luò)請(qǐng)求模塊
• pip install aiohttp
• 使用代碼：
• 1，首先寫出一個(gè)大致架構(gòu)

async def get_request(url):with aiohttp.ClientSession() as sess:#實(shí)例化一個(gè)請(qǐng)求對(duì)象# get/post(url,headers,params/data,proxy="http://ip:port")with sess.get(url=url) as response:# 使用get發(fā)起請(qǐng)求，返回一個(gè)相應(yīng)對(duì)象page_text = response.text#獲取字符串形式的相應(yīng)數(shù)據(jù)return page_text

• 2，補(bǔ)充細(xì)節(jié)
  • 在阻塞操作前加上await關(guān)鍵字
  • 在每一個(gè)with前加上async關(guān)鍵字

async def get_request(url):async with aiohttp.ClientSession() as sess:#實(shí)例化一個(gè)請(qǐng)求對(duì)象# get/post(url,headers,params/data,proxy="http://ip:port")async with await sess.get(url=url) as response:# 使用get發(fā)起請(qǐng)求，返回一個(gè)相應(yīng)對(duì)象page_text = await response.text#text()獲取字符串形式的相應(yīng)數(shù)據(jù) read()獲取byte類型的響應(yīng)數(shù)據(jù)return page_text

完整爬蟲(chóng)異步網(wǎng)站如下：

import asyncio import aiohttp import timeurls = ["http://127.0.0.1:5000/gpc","http://127.0.0.1:5000/una","http://127.0.0.1:5000/python" ]async def get_request(url):async with aiohttp.ClientSession() as sess:#實(shí)例化一個(gè)請(qǐng)求對(duì)象# get/post(url,headers,params/data,proxy="http://ip:port")async with await sess.get(url=url) as response:# 使用get發(fā)起請(qǐng)求，返回一個(gè)相應(yīng)對(duì)象page_text = await response.text()#text()獲取字符串形式的相應(yīng)數(shù)據(jù) read()獲取byte類型的響應(yīng)數(shù)據(jù)return page_textif __name__=="__main__":start = time.time()tasks = [] # 多任務(wù)列表# 1.創(chuàng)建協(xié)程對(duì)象for url in urls:c = get_request(url)# 2.創(chuàng)建任務(wù)對(duì)象task = asyncio.ensure_future(c)tasks.append(task)# 3.創(chuàng)建事件循環(huán)對(duì)象loop = asyncio.get_event_loop()# 4.將任務(wù)對(duì)象注冊(cè)到事件循環(huán)中且開(kāi)啟事件循環(huán)loop.run_until_complete(asyncio.wait(tasks))# 必須使用wait方法對(duì)tasks封裝print("總耗時(shí):",time.time()-start)

多任務(wù)爬蟲(chóng)的數(shù)據(jù)解析

• 一定要是用任務(wù)對(duì)象的回調(diào)函數(shù)實(shí)現(xiàn)數(shù)據(jù)解析
• 多任務(wù)的架構(gòu)中數(shù)據(jù)的爬取是封裝在特殊函數(shù)中的，我們一定要保證數(shù)據(jù)請(qǐng)求結(jié)束后，在實(shí)現(xiàn)數(shù)據(jù)解析
• 使用多任務(wù)的異步協(xié)程爬取數(shù)據(jù)實(shí)現(xiàn)套路：
  • 可以先試用request模塊將待請(qǐng)求的數(shù)據(jù)對(duì)應(yīng)的url封裝到urls列表中【同步】
  • 可以使用aiohttp模塊將列表中的url進(jìn)行異步的請(qǐng)求和數(shù)據(jù)解析【異步】
    代碼展示

import asyncio import aiohttp import time from lxml import etreeurls = ["http://127.0.0.1:5000/gpc","http://127.0.0.1:5000/una","http://127.0.0.1:5000/python" ]async def get_request(url):async with aiohttp.ClientSession() as sess:#實(shí)例化一個(gè)請(qǐng)求對(duì)象# get/post(url,headers,params/data,proxy="http://ip:port")async with await sess.get(url=url) as response:# 使用get發(fā)起請(qǐng)求，返回一個(gè)相應(yīng)對(duì)象page_text = await response.text()#text()獲取字符串形式的相應(yīng)數(shù)據(jù) read()獲取byte類型的響應(yīng)數(shù)據(jù)return page_text #解析函數(shù)的封裝 def parse(t):#獲取請(qǐng)求到頁(yè)面源碼數(shù)據(jù)page_text = t.result()tree = etree.HTML(page_text)parse_text = tree.xpath("//div[@id='contson09d31b73b44d']/p[5]//text()")print(parse_text) if __name__=="__main__":start = time.time()tasks = [] # 多任務(wù)列表# 1.創(chuàng)建協(xié)程對(duì)象for url in urls:c = get_request(url)# 2.創(chuàng)建任務(wù)對(duì)象task = asyncio.ensure_future(c)task.add_done_callback(parse)tasks.append(task)# 3.創(chuàng)建事件循環(huán)對(duì)象loop = asyncio.get_event_loop()# 4.將任務(wù)對(duì)象注冊(cè)到事件循環(huán)中且開(kāi)啟事件循環(huán)loop.run_until_complete(asyncio.wait(tasks))# 必須使用wait方法對(duì)tasks封裝print("總耗時(shí):",time.time()-start)

總結(jié)

以上是生活随笔為你收集整理的10-异步爬虫（线程池/asyncio协程）实战案例的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。