Python: 基於線程池的異步/多任務異步協程 【asyncio】

同步代碼

import time
import requests

urls = [
    'http://www.chinadaily.com.cn/',
    'https://www.163.com',
    'https://www.bilibili.com/'
]
def get_request(url):
    page_text = requests.get(url).text
    print(len(page_text))

if __name__ == "__main__":
    start = time.time()
    for url in urls:
        get_request(url)
    print('總耗時：',time.time()-start)

## 總耗時： 1.0235168933868408

基於線程池的異步效果

import time
from multiprocessing.dummy import Pool
import requests

urls = [
    'http://www.chinadaily.com.cn/',
    'https://www.163.com',
    'https://www.bilibili.com/'
]
def get_request(url):
    page_text = requests.get(url).text
    return len(page_text)

if __name__ == "__main__":
    start = time.time()

    pool = Pool(3) #啟動了三個線程
    #參數1：回調函數
    #參數2：可迭代的對象，alist
    #作用：可以將alist中的每一個元素依次傳遞給回調函數作為參數，然后回調函數會異步
        #對列表中的元素進行相關操作運算
    #map的返回值就是回調函數返回的所有結果
    page_text_len_list = pool.map(get_request,urls)
    print(page_text_len_list)
    print('總耗時：',time.time()-start)

## 總耗時： 0.5089230537414551

多任務異步協程 【asyncio】

import requests
import asyncio
import time
import aiohttp
from lxml import etree

# - 特殊函數
#     - 如果一個函數的定義被async關鍵字修飾，則該函數就編程了一個特殊的函數
# - 特殊之處：
#     - 該函數調用后函數內部的實現語句不會被【立即】執行
#     - 該函數被調用后會返回一個協程對象
# 特殊的函數：不可以出現不支持異步模塊的代碼，不可以使用requests模塊
async def get_request(url):
    # 使用aiohttp進行網絡請求
    # - aiohttp的編碼使用：
    #     - 編寫一個大致的架構
    #         with aiohttp.ClientSession() as sess:
    #             with sess.get(url=url) as response:
    #                 page_text = response.text()
    #                 return page_text
    #     - 在架構中補充細節
    #         - 在每一個with前加上async關鍵字
    #         - 在每一個阻塞操作前加上一個await關鍵字
    #     - 完整代碼：
    #         async with aiohttp.ClientSession() as sess:
    #             async with await sess.get(url=url) as response:
    #                 page_text = await response.text()
    #                 return page_text
    #     - await關鍵字可以確保在異步執行操作的過程中可以保證阻塞操作執行完畢。
    async with aiohttp.ClientSession() as sess: # 實例化一個請求對象叫做sess
        # sess.get(url,headers,params,proxy)
        # sess.post(url,headers,data,proxy)
        # proxy參數的用法和requests不一樣，剩下參數和requests的用法一樣
        # proxy = "http://ip:port"
        async with await sess.get(url=url) as response: # 調用get發請求，返回一個響應對象
            # text()返回字符串形式的響應數據
            # read()返回bytes類型的響應數據
            page_text = await response.text() # 獲取了頁面源碼數據
            return page_text

# 定義一個任務對象的回調函數
# 注意：回調函數必須要有一個參數，該參數表示就是該函數的綁定者
# 多任務的異步爬蟲中數據解析或者持久化存儲的操作需要寫在任務對象的回調函數中
def parse(task):
    #result():返回的就是特殊函數的返回值
    page_text = task.result()
    len_page_text = len(page_text)
    print(len_page_text)

if __name__ == "__main__":
    start = time.time()
    urls = [
        'http://www.chinadaily.com.cn/',
        'https://www.163.com',
        'https://www.bilibili.com/'
    ]
    # 定義一個任務列表
    tasks = []
    for url in urls:
        # 創建三個協程對象
        # - 協程對象: 該對象是特殊函數調用后返回的協程對象。
        # - 協程對象 == 特殊的函數
        # - 一個函數表示指定的一組操作，那么一個協程對象表示一組特定的操作
        # - 如何創建一個協程對象？- 調用特殊函數返回
        c = get_request(url)
        #創建三個任務對象
        # - 任務對象 - 就是一個高級的協程對象。
        # - 任務對象 == 協程對象 == 特殊的函數
        # - 任務對象也表示一組特定的操作
        # - 如何創建一個任務對象？- task = asyncio.ensure_future(c)  # c指的是協程對象
        task = asyncio.ensure_future(c)
        # - 問題1：如何獲取特殊函數中的返回值？
        #     - 基於任務對象的回調函數。
        #     - 如何給任務對象綁定回調函數？
        #         - task.add_done_callback(parse)
        #         - parse就是回調函數
        #             - parse必須要有一個參數，參數就是該函數的綁定者(任務對象)
        #             - result()返回的就是特殊函數的返回值
        task.add_done_callback(parse) # 綁定回調
        tasks.append(task)

    # - 事件循環：EventLoop
    #     - 對象。
    #     - 如何創建該對象 - loop = asyncio.get_event_loop()
    #     - 作用：
    #         - 是用來裝載任務（協程）對象的：可以將事件循環當做是一個容器，容器中存放的是多個任務對象
    #         - 如果事件循環存放了多個任務對象且事件循環啟動后，則事件循環對象就可以異步的將每一個任務對象對應的指定操作進行執行。
    #     - 如何將任務對象存儲且啟動事件循環對象
    #         - loop.run_until_complete(task)#將一個任務對象進行了存儲
    loop = asyncio.get_event_loop()
    #將任務列表中的多個任務注冊到了事件循環中
    # - 問題2：任務對象在事件循環中是否被異步的執行？
    #     - 一個任務對象在事件循環中無法檢測出是否被異步執行
    #     - 需要將多個任務對象注冊事件循環中進行測試：
    #         - 如何將多個任務注冊到事件循環中？
    #         - loop.run_until_complete(asyncio.wait(tasks))  # tasks表示的是任務列表
    # - wait()方法的作用：
    #     - 表示掛起的意思。
    #     - asyncio.wait(tasks）將任務列表中的每一個任務對象進行掛起。
    #     - 掛起：讓當前的任務對象交出cpu的使用權。
    loop.run_until_complete(asyncio.wait(tasks))
    print('總耗時：',time.time()-start)

## 總耗時： 0.49843263626098633

 

- 實戰說明
- 如果想使用該模式進行異步的數據爬取則必須：
- 將等待即將被爬取的頁面的url單獨的抽取存儲到一個列表中。
- 通常情況下的玩法：
- 使用requests將等待爬取頁面的url獲取
- 將url寫入列表，使用多任務異步協程爬取列表中的頁面數據

參考：https://www.bilibili.com/video/BV1tE411F7do（P13~P15）