python?基于aiohttp的異步爬蟲(chóng)實(shí)戰(zhàn)詳解

更新時(shí)間：2022年09月09日 16:08:02 作者：鋼鐵俠的知識(shí)庫(kù)

這篇文章主要為大家介紹了python?基于aiohttp的異步爬蟲(chóng)實(shí)戰(zhàn)詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

引言

鋼鐵知識(shí)庫(kù)，一個(gè)學(xué)習(xí)python爬蟲(chóng)、數(shù)據(jù)分析的知識(shí)庫(kù)。人生苦短，快用python。

之前我們使用requests庫(kù)爬取某個(gè)站點(diǎn)的時(shí)候，每發(fā)出一個(gè)請(qǐng)求，程序必須等待網(wǎng)站返回響應(yīng)才能接著運(yùn)行，而在整個(gè)爬蟲(chóng)過(guò)程中，整個(gè)爬蟲(chóng)程序是一直在等待的，實(shí)際上沒(méi)有做任何事情。

像這種占用磁盤(pán)/內(nèi)存IO、網(wǎng)絡(luò)IO的任務(wù)，大部分時(shí)間是CPU在等待的操作，就叫IO密集型任務(wù)。對(duì)于這種情況有沒(méi)有優(yōu)化方案呢，當(dāng)然有，那就是使用aiohttp庫(kù)實(shí)現(xiàn)異步爬蟲(chóng)。

aiohttp是什么

我們?cè)谑褂胷equests請(qǐng)求時(shí)，只能等一個(gè)請(qǐng)求先出去再回來(lái)，才會(huì)發(fā)送下一個(gè)請(qǐng)求。明顯效率不高阿，這時(shí)候如果換成異步請(qǐng)求的方式，就不會(huì)有這個(gè)等待。一個(gè)請(qǐng)求發(fā)出去，不管這個(gè)請(qǐng)求什么時(shí)間響應(yīng)，程序通過(guò)await掛起協(xié)程對(duì)象后直接進(jìn)行下一個(gè)請(qǐng)求。

解決方法就是通過(guò) aiohttp + asyncio，什么是aiohttp？一個(gè)基于 asyncio 的異步 HTTP 網(wǎng)絡(luò)模塊，可用于實(shí)現(xiàn)異步爬蟲(chóng)，速度明顯快于 requests 的同步爬蟲(chóng)。

requests和aiohttp區(qū)別

區(qū)別就是一個(gè)同步一個(gè)是異步。話不多說(shuō)直接上代碼看效果。

安裝aiohttp

pip install aiohttp

requests同步示例：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# author: 鋼鐵知識(shí)庫(kù)
import time
import requests
# 同步請(qǐng)求
def main():
    start = time.time()
    for i in range(5):
        res = requests.get('http://httpbin.org/delay/2')
        print(f'當(dāng)前時(shí)間：{datetime.datetime.now()}, status_code = {res.status_code}')
    print(f'requests同步耗時(shí)：{time.time() - start}')
if __name__ == '__main__':
    main()
'''
當(dāng)前時(shí)間：2022-09-05 15:44:51.991685, status_code = 200
當(dāng)前時(shí)間：2022-09-05 15:44:54.528918, status_code = 200
當(dāng)前時(shí)間：2022-09-05 15:44:57.057373, status_code = 200
當(dāng)前時(shí)間：2022-09-05 15:44:59.643119, status_code = 200
當(dāng)前時(shí)間：2022-09-05 15:45:02.167362, status_code = 200
requests同步耗時(shí)：12.785893440246582
'''

可以看到5次請(qǐng)求總共用12.7秒，再來(lái)看同樣的請(qǐng)求異步多少時(shí)間。

aiohttp異步示例：

#!/usr/bin/env python
# file: day6-9同步和異步.py
# author: 鋼鐵知識(shí)庫(kù)
import asyncio
import time
import aiohttp
async def async_http():
    # 聲明一個(gè)支持異步的上下文管理器
    async with aiohttp.ClientSession() as session:
        res = await session.get('http://httpbin.org/delay/2')
        print(f'當(dāng)前時(shí)間：{datetime.datetime.now()}, status_code = {res.status}')
tasks = [async_http() for _ in range(5)]
start = time.time()
# Python 3.7 及以后,不需要顯式聲明事件循環(huán),可以使用 asyncio.run()來(lái)代替最后的啟動(dòng)操作
asyncio.run(asyncio.wait(tasks))
print(f'aiohttp異步耗時(shí)：{time.time() - start}')
'''
當(dāng)前時(shí)間：2022-09-05 15:42:32.363966, status_code = 200
當(dāng)前時(shí)間：2022-09-05 15:42:32.366957, status_code = 200
當(dāng)前時(shí)間：2022-09-05 15:42:32.374973, status_code = 200
當(dāng)前時(shí)間：2022-09-05 15:42:32.384909, status_code = 200
當(dāng)前時(shí)間：2022-09-05 15:42:32.390318, status_code = 200
aiohttp異步耗時(shí)：2.5826876163482666
'''

兩次對(duì)比可以看到執(zhí)行過(guò)程，時(shí)間一個(gè)是順序執(zhí)行，一個(gè)是同時(shí)執(zhí)行。這就是同步和異步的區(qū)別。

aiohttp使用介紹

接下來(lái)我們會(huì)詳細(xì)介紹aiohttp庫(kù)的用法和爬取實(shí)戰(zhàn)。aiohttp 是一個(gè)支持異步請(qǐng)求的庫(kù)，它和 asyncio 配合使用，可以使我們非常方便地實(shí)現(xiàn)異步請(qǐng)求操作。asyncio模塊，其內(nèi)部實(shí)現(xiàn)了對(duì)TCP、UDP、SSL協(xié)議的異步操作，但是對(duì)于HTTP請(qǐng)求，就需要aiohttp實(shí)現(xiàn)了。

aiohttp分為兩部分，一部分是Client，一部分是Server。下面來(lái)說(shuō)說(shuō)aiohttp客戶端部分的用法。

基本實(shí)例

先寫(xiě)一個(gè)簡(jiǎn)單的案例

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Author  : 鋼鐵知識(shí)庫(kù)
import asyncio
import aiohttp
async def get_api(session, url):
    # 聲明一個(gè)支持異步的上下文管理器
    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 get_api(session, 'http://httpbin.org/delay/2')
        print(f'html: {html[:50]}')
        print(f'status : {status}')
if __name__ == '__main__':
    #  Python 3.7 及以后,不需要顯式聲明事件循環(huán),可以使用 asyncio.run(main())來(lái)代替最后的啟動(dòng)操作
    asyncio.get_event_loop().run_until_complete(main())
'''
html: {
  "args": {}, 
  "data": "", 
  "files": {}, 
status : 200
Process finished with exit code 0
'''

aiohttp請(qǐng)求的方法和之前有明顯區(qū)別，主要包括如下幾點(diǎn)：

除了導(dǎo)入aiohttp庫(kù)，還必須引入asyncio庫(kù)，因?yàn)橐獙?shí)現(xiàn)異步，需要啟動(dòng)協(xié)程。
異步的方法定義不同，前面都要統(tǒng)一加async來(lái)修飾。
with as用于聲明上下文管理器，幫我們自動(dòng)分配和釋放資源，加上async代碼支持異步。
對(duì)于返回協(xié)程對(duì)象的操作，前面需要加await來(lái)修飾。response.text()返回的是協(xié)程對(duì)象。
最后運(yùn)行啟用循環(huán)事件

注意：Python3.7及以后的版本中，可以使用asyncio.run(main())代替最后的啟動(dòng)操作。

URL參數(shù)設(shè)置

對(duì)于URL參數(shù)的設(shè)置，我們可以借助params設(shè)置，傳入一個(gè)字典即可，實(shí)例如下：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Author  : 鋼鐵知識(shí)庫(kù)
import aiohttp
import asyncio
async def main():
    params = {'name': '鋼鐵知識(shí)庫(kù)', 'age': 23}
    async with aiohttp.ClientSession() as session:
        async with session.get('https://www.httpbin.org/get', params=params) as res:
            print(await res.json())
if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())
'''
{'args': {'age': '23', 'name': '鋼鐵知識(shí)庫(kù)'}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Host': 'www.httpbin.org', 'User-Agent': 'Python/3.8 aiohttp/3.8.1', 'X-Amzn-Trace-Id': 'Root=1-63162e34-1acf7bde7a6d801368494c72'}, 'origin': '122.55.11.188', 'url': 'https://www.httpbin.org/get?name=鋼鐵知識(shí)庫(kù)&amp;age=23'}
'''

可以看到實(shí)際請(qǐng)求的URL后面帶了后綴，這就是params的內(nèi)容。

請(qǐng)求類型

除了get請(qǐng)求，aiohttp還支持其它請(qǐng)求類型，如POST、PUT、DELETE等，和requests使用方式類似。

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

要使用這些方法，只需要把對(duì)應(yīng)的方法和參數(shù)替換一下。用法和get類似就不再舉例。

響應(yīng)的幾個(gè)方法

對(duì)于響應(yīng)來(lái)說(shuō)，我們可以用如下方法分別獲取其中的響應(yīng)情況。狀態(tài)碼、響應(yīng)頭、響應(yīng)體、響應(yīng)體二進(jìn)制內(nèi)容、響應(yīng)體JSON結(jié)果，實(shí)例如下：

#!/usr/bin/env python
# @Author  : 鋼鐵知識(shí)庫(kù)
import aiohttp
import asyncio
async def main():
    data = {'name': '鋼鐵知識(shí)庫(kù)', 'age': 23}
    async with aiohttp.ClientSession() as session:
        async with session.post('https://www.httpbin.org/post', data=data) as response:
            print('status:', response.status)  # 狀態(tài)碼
            print('headers:', response.headers)  # 響應(yīng)頭
            print('body:', await response.text())  # 響應(yīng)體
            print('bytes:', await response.read())  # 響應(yīng)體二進(jìn)制內(nèi)容
            print('json:', await response.json())  # 響應(yīng)體json數(shù)據(jù)
if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())

'''
status: 200
headers: &lt;CIMultiDictProxy('Date': 'Tue, 06 Sep 2022 00:18:36 GMT', 'Content-Type': 'application/json', 'Content-Length': '534', 'Connection': 'keep-alive', 'Server': 'gunicorn/19.9.0', 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true')&gt;
body: {
  "args": {}, 
  "data": "", 
  "files": {}, 
  "form": {
    "age": "23", 
    "name": "\u94a2\u94c1\u77e5\u8bc6\u5e93"
  }, 
  "headers": {
    "Accept": "*/*", 
    "Accept-Encoding": "gzip, deflate", 
    "Content-Length": "57", 
    "Content-Type": "application/x-www-form-urlencoded", 
    "Host": "www.httpbin.org", 
    "User-Agent": "Python/3.8 aiohttp/3.8.1", 
    "X-Amzn-Trace-Id": "Root=1-631691dc-6aa1b2b85045a1a0481d06e1"
  }, 
  "json": null, 
  "origin": "122.55.11.188", 
  "url": "https://www.httpbin.org/post"
}
bytes: b'{\n  "args": {}, \n  "data": "", \n  "files": {}, \n  "form": {\n    "age": "23", \n    "name": "\\u94a2\\u94c1\\u77e5\\u8bc6\\u5e93"\n  }, \n  "headers": {\n    "Accept": "*/*", \n    "Accept-Encoding": "gzip, deflate", \n    "Content-Length": "57", \n    "Content-Type": "application/x-www-form-urlencoded", \n    "Host": "www.httpbin.org", \n    "User-Agent": "Python/3.8 aiohttp/3.8.1", \n    "X-Amzn-Trace-Id": "Root=1-631691dc-6aa1b2b85045a1a0481d06e1"\n  }, \n  "json": null, \n  "origin": "122.5.132.196", \n  "url": "https://www.httpbin.org/post"\n}\n'
json: {'args': {}, 'data': '', 'files': {}, 'form': {'age': '23', 'name': '鋼鐵知識(shí)庫(kù)'}, 'headers': {'Accept': '*/*', 'Accept-Encoding': 'gzip, deflate', 'Content-Length': '57', 'Content-Type': 'application/x-www-form-urlencoded', 'Host': 'www.httpbin.org', 'User-Agent': 'Python/3.8 aiohttp/3.8.1', 'X-Amzn-Trace-Id': 'Root=1-631691dc-6aa1b2b85045a1a0481d06e1'}, 'json': None, 'origin': '122.55.11.188', 'url': 'https://www.httpbin.org/post'}
'''

可以看到有些字段前面需要加await，因?yàn)槠浞祷氐氖且粋€(gè)協(xié)程對(duì)象(如async修飾的方法)，那么前面就要加await。

超時(shí)設(shè)置

我們可以借助ClientTimeout對(duì)象設(shè)置超時(shí)，例如要設(shè)置1秒的超時(shí)時(shí)間，可以這么實(shí)現(xiàn)：

#!/usr/bin/env python
# @Author  : 鋼鐵知識(shí)庫(kù)
import aiohttp
import asyncio
async def main():
    # 設(shè)置 1 秒的超時(shí) 
    timeout = aiohttp.ClientTimeout(total=1)
    data = {'name': '鋼鐵知識(shí)庫(kù)', 'age': 23}
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.get('https://www.httpbin.org/delay/2', data=data) as response:
            print('status:', response.status)  # 狀態(tài)碼
if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())
'''
Traceback (most recent call last):
####中間省略####
    raise asyncio.TimeoutError from None
asyncio.exceptions.TimeoutError
'''

這里設(shè)置了超時(shí)1秒請(qǐng)求延時(shí)2秒，發(fā)現(xiàn)拋出異常asyncio.TimeoutError，如果正常則響應(yīng)200。

并發(fā)限制

aiohttp可以支持非常高的并發(fā)量，但面對(duì)高并發(fā)網(wǎng)站可能會(huì)承受不住，隨時(shí)有掛掉的危險(xiǎn)，這時(shí)需要對(duì)并發(fā)進(jìn)行一些控制?，F(xiàn)在我們借助asyncio 的Semaphore來(lái)控制并發(fā)量，實(shí)例如下：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Author  : 鋼鐵知識(shí)庫(kù)
import asyncio
from datetime import datetime
import aiohttp
# 聲明最大并發(fā)量
semaphore = asyncio.Semaphore(2)
async def get_api():
    async with semaphore:
        print(f'scrapting...{datetime.now()}')
        async with session.get('https://www.baidu.com') as response:
            await asyncio.sleep(2)
            # print(f'當(dāng)前時(shí)間：{datetime.now()}, {response.status}')
async def main():
    global session
    session = aiohttp.ClientSession()
    tasks = [asyncio.ensure_future(get_api()) for _ in range(1000)]
    await asyncio.gather(*tasks)
    await session.close()
if __name__ == '__main__':
    asyncio.get_event_loop().run_until_complete(main())
'''
scrapting...2022-09-07 08:11:14.190000
scrapting...2022-09-07 08:11:14.292000
scrapting...2022-09-07 08:11:16.482000
scrapting...2022-09-07 08:11:16.504000
scrapting...2022-09-07 08:11:18.520000
scrapting...2022-09-07 08:11:18.521000
'''

在main方法里，我們聲明了1000個(gè)task，如果沒(méi)有通過(guò)Semaphore進(jìn)行并發(fā)限制，那這1000放到gather方法后會(huì)被同時(shí)執(zhí)行，并發(fā)量相當(dāng)大。有了信號(hào)量的控制之后，同時(shí)運(yùn)行的task數(shù)量就會(huì)被控制，這樣就能給aiohttp限制速度了。

aiohttp異步爬取實(shí)戰(zhàn)

接下來(lái)我們通過(guò)異步方式練手一個(gè)小說(shuō)爬蟲(chóng)，需求如下：

需求頁(yè)面：https://dushu.baidu.com/pc/detail?gid=4308080950

目錄接口：https://dushu.baidu.com/api/pc/getCatalog?data={"book_id":"4308080950"}

詳情接口：

https://dushu.baidu.com/api/pc/getChapterContent?data={"book_id":"4295122774","cid":"4295122774|116332"}

關(guān)鍵參數(shù)：book_id：小說(shuō)ID、cid：章節(jié)id

采集要求：使用協(xié)程方式寫(xiě)入，數(shù)據(jù)存放進(jìn)mongo

需求分析：點(diǎn)開(kāi)需求頁(yè)面，通過(guò)F12抓包可以發(fā)現(xiàn)兩個(gè)接口。一個(gè)目錄接口，一個(gè)詳情接口。
首先第一步先請(qǐng)求目錄接口拿到cid章節(jié)id，然后將cid傳遞給詳情接口拿到小說(shuō)數(shù)據(jù)，最后存入mongo即可。

話不多說(shuō)，直接上代碼：

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Author  : 鋼鐵知識(shí)庫(kù)
# 不合適就是不合適，真正合適的，你不會(huì)有半點(diǎn)猶豫。
import asyncio
import json,re
import logging
import aiohttp
import requests
from utils.conn_db import ConnDb
# 日志格式
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s: %(message)s')
# 章節(jié)目錄api
b_id = '4308080950'
url = 'https://dushu.baidu.com/api/pc/getCatalog?data={"book_id":"'+b_id+'"}'
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) "
                  "Chrome/104.0.0.0 Safari/537.36"
}
# 并發(fā)聲明
semaphore = asyncio.Semaphore(5)
async def download(title,b_id, cid):
    data = {
        "book_id": b_id,
        "cid": f'{b_id}|{cid}',
    }
    data = json.dumps(data)
    detail_url = 'https://dushu.baidu.com/api/pc/getChapterContent?data={}'.format(data)
    async with semaphore:
        async with aiohttp.ClientSession(headers=headers) as session:
            async with session.get(detail_url) as response:
                res = await response.json()
                content = {
                    'title': title,
                    'content': res['data']['novel']['content']
                }
                # print(title)
                await save_data(content)
async def save_data(data):
    if data:
        client = ConnDb().conn_motor_mongo()
        db = client.baidu_novel
        collection = db.novel
        logging.info('saving data %s', data)
        await collection.update_one(
            {'title': data.get('title')},
            {'$set': data},
            upsert=True
        )
async def main():
    res = requests.get(url, headers=headers)
    tasks = []
    for re in res.json()['data']['novel']['items']:     # 拿到某小說(shuō)目錄cid
        title = re['title']
        cid = re['cid']
        tasks.append(download(title, b_id, cid))    # 將請(qǐng)求放到列表里，再通過(guò)gather執(zhí)行并發(fā)
    await asyncio.gather(*tasks)
if __name__ == '__main__':
    asyncio.run(main())

至此，我們就使用aiohttp完成了對(duì)小說(shuō)章節(jié)的爬取。

要實(shí)現(xiàn)異步處理，得先要有掛起操作，當(dāng)一個(gè)任務(wù)需要等待 IO 結(jié)果的時(shí)候，可以掛起當(dāng)前任務(wù)，轉(zhuǎn)而去執(zhí)行其他任務(wù)，這樣才能充分利用好資源，要實(shí)現(xiàn)異步，需要了解 await 的用法，使用 await 可以將耗時(shí)等待的操作掛起，讓出控制權(quán)。當(dāng)協(xié)程執(zhí)行的時(shí)候遇到 await，時(shí)間循環(huán)就會(huì)將本協(xié)程掛起，轉(zhuǎn)而去執(zhí)行別的協(xié)程，直到其他的協(xié)程掛起或執(zhí)行完畢。

await 后面的對(duì)象必須是如下格式之一：

A native coroutine object returned from a native coroutine function，一個(gè)原生 coroutine 對(duì)象。
A generator-based coroutine object returned from a function decorated with types.coroutine，一個(gè)由 types.coroutine 修飾的生成器，這個(gè)生成器可以返回 coroutine 對(duì)象。
An object with an await method returning an iterator，一個(gè)包含 await 方法的對(duì)象返回的一個(gè)迭代器。