Python爬蟲進階 | 異步協程

一、背景

　　之前爬蟲使用的是requests+多線程/多進程，后來隨着前幾天的深入了解，才發現，對於爬蟲來說，真正的瓶頸並不是CPU的處理速度，而是對於網頁抓取時候的往返時間，因為如果采用requests+多線程/多進程，他本身是阻塞式的編程，所以時間都花費在了等待網頁結果的返回和對爬取到的數據的寫入上面。而如果采用非阻塞編程，那么就沒有這個困擾。這邊首先要理解一下阻塞和非阻塞的區別。

（1）阻塞調用是指調用結果返回之前，當前線程會被掛起（線程進入非可執行狀態，在這個狀態下，CPU不會給線程分配時間片，即線程暫停運行）。函數只有在得到結果之后才會返回。

（2）對於非阻塞則不會掛起，直接執行接下去的程序，返回結果后再回來處理返回值。

 

　　其實爬蟲的本質就是client發請求，批量獲取server的響應數據，如果我們有多個url待爬取，只用一個線程且采用串行的方式執行，那只能等待爬取一個結束后才能繼續下一個，效率會非常低。需要強調的是：對於單線程下串行N個任務，並不完全等同於低效，如果這N個任務都是純計算的任務，那么該線程對cpu的利用率仍然會很高，之所以單線程下串行多個爬蟲任務低效，是因為爬蟲任務是明顯的IO密集型（阻塞）程序。那么該如何提高爬取性能呢？

 

二、基本概念

2.1 阻塞

阻塞狀態指程序未得到所需計算資源時被掛起的狀態。程序在等待某個操作完成期間，自身無法繼續干別的事情，則稱該程序在該操作上是阻塞的。

常見的阻塞形式有：網絡 I/O 阻塞、磁盤 I/O 阻塞、用戶輸入阻塞等。阻塞是無處不在的，包括 CPU 切換上下文時，所有的進程都無法真正干事情，它們也會被阻塞。如果是多核 CPU 則正在執行上下文切換操作的核不可被利用。

 

2.2 非阻塞

程序在等待某操作過程中，自身不被阻塞，可以繼續運行干別的事情，則稱該程序在該操作上是非阻塞的。

非阻塞並不是在任何程序級別、任何情況下都可以存在的。僅當程序封裝的級別可以囊括獨立的子程序單元時，它才可能存在非阻塞狀態。

非阻塞的存在是因為阻塞存在，正因為某個操作阻塞導致的耗時與效率低下，我們才要把它變成非阻塞的。

 

2.3 同步

不同程序單元為了完成某個任務，在執行過程中需靠某種通信方式以協調一致，稱這些程序單元是同步執行的。例如購物系統中更新商品庫存，需要用“行鎖”作為通信信號，讓不同的更新請求強制排隊順序執行，那更新庫存的操作是同步的。簡言之，同步意味着有序。

 

2.4 異步

為完成某個任務，不同程序單元之間過程中無需通信協調，也能完成任務的方式，不相關的程序單元之間可以是異步的。例如，爬蟲下載網頁。調度程序調用下載程序后，即可調度其他任務，而無需與該下載任務保持通信以協調行為。不同網頁的下載、保存等操作都是無關的，也無需相互通知協調。這些異步操作的完成時刻並不確定。簡言之，異步意味着無序。

 

2.5 多進程

多進程就是利用 CPU 的多核優勢，在同一時間並行地執行多個任務，可以大大提高執行效率。

 

2.6 協程

協程，英文叫做 Coroutine，又稱微線程，纖程，協程是一種用戶態的輕量級線程。

協程擁有自己的寄存器上下文和棧。協程調度切換時，將寄存器上下文和棧保存到其他地方，在切回來的時候，恢復先前保存的寄存器上下文和棧。因此協程能保留上一次調用時的狀態，即所有局部狀態的一個特定組合，每次過程重入時，就相當於進入上一次調用的狀態。

協程本質上是個單進程，協程相對於多進程來說，無需線程上下文切換的開銷，無需原子操作鎖定及同步的開銷，編程模型也非常簡單。

我們可以使用協程來實現異步操作，比如在網絡爬蟲場景下，我們發出一個請求之后，需要等待一定的時間才能得到響應，但其實在這個等待過程中，程序可以干許多其他的事情，等到響應得到之后才切換回來繼續處理，這樣可以充分利用 CPU 和其他資源，這就是異步協程的優勢。

 

三、分析處理 

　　同步調用：即提交一個任務后就在原地等待任務結束，等到拿到任務的結果后再繼續下一行代碼，效率低

import requests

def get_page(url):
    print('下載 %s' %url)
    response=requests.get(url)
    if response.status_code == 200:
        return response.text

def parse_page(res):
    print('解析 %s' %(len(res)))


def main():
    urls=['https://www.baidu.com/','http://www.sina.com.cn/','https://www.python.org']
    for url in urls:
        res=get_page(url)                         #調用一個任務，就在原地等待任務結束拿到結果后才繼續往后執行
        parse_page(res)

if __name__ == "__main__":
    main()

a. 解決同步調用方案之多線程/多進程

好處：在服務器端使用多線程（或多進程）。多線程（或多進程）的目的是讓每個連接都擁有獨立的線程（或進程），這樣任何一個連接的阻塞都不會影響其他的連接。

弊端：開啟多進程或都線程的方式，我們是無法無限制地開啟多進程或多線程的：在遇到要同時響應成百上千路的連接請求，則無論多線程還是多進程都會嚴重占據系統資源，降低系統對外界響應效率，而且線程與進程本身也更容易進入假死狀態。

b. 解決同步調用方案之線程/進程池

好處：很多程序員可能會考慮使用“線程池”或“連接池”。“線程池”旨在減少創建和銷毀線程的頻率，其維持一定合理數量的線程，並讓空閑的線程重新承擔新的執行任務。可以很好的降低系統開銷。

弊端：“線程池”和“連接池”技術也只是在一定程度上緩解了頻繁調用IO接口帶來的資源占用。而且，所謂“池”始終有其上限，當請求大大超過上限時，“池”構成的系統對外界的響應並不比沒有池的時候效果好多少。所以使用“池”必須考慮其面臨的響應規模，並根據響應規模調整“池”的大小。

 

案例：基於multiprocessing.dummy線程池爬取梨視頻的視頻信息

import requests
import re
from lxml import etree
from multiprocessing.dummy import Pool
import random

header = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.119 Safari/537.36'
}

def get_page(url):
    response = requests.get(url=url,headers=header)
    if response.status_code == 200:
        return response.text
    return None

def parse_page(res):
    tree = etree.HTML(res)
    li_list = tree.xpath('//div[@id="listvideoList"]/ul/li')

    video_url_list = []
    for li in li_list:
        detail_url = 'https://www.pearvideo.com/' + li.xpath('./div/a/@href')[0]
        detail_page = requests.get(url=detail_url, headers=header).text
        video_url = re.findall('srcUrl="(.*?)",vdoUrl', detail_page, re.S)[0]
        video_url_list.append(video_url)

    return video_url_list

# 獲取視頻
def getVideoData(url):
    return requests.get(url=url, headers=header).content

# 持久化存儲
def saveVideo(data):
    fileName = str(random.randint(0, 5000)) + '.mp4'  # 因回調函數只能傳一個參數，所以沒辦法再傳名字了，只能自己取名
    with open(fileName, 'wb') as fp:
        fp.write(data)

def main():
    url = 'https://www.pearvideo.com/category_1'
    res = get_page(url)
    links = parse_page(res)


    pool = Pool(5)  # 實例化一個線程池對象

    #  pool.map(回調函數，可迭代對象）函數依次執行對象
    video_data_list = pool.map(getVideoData, links)
    pool.map(saveVideo, video_data_list)
    
 pool.close() pool.join() if __name__== "__main__":
    main()

總結：對應上例中的所面臨的可能同時出現的上千甚至上萬次的客戶端請求，“線程池”或“連接池”或許可以緩解部分壓力，但是不能解決所有問題。總之，多線程模型可以方便高效的解決小規模的服務請求，但面對大規模的服務請求，多線程模型也會遇到瓶頸，可以用非阻塞接口來嘗試解決這個問題。

 

終極處理方案

　　上述無論哪種方案都沒有解決一個性能相關的問題：IO阻塞，無論是多進程還是多線程，在遇到IO阻塞時都會被操作系統強行剝奪走CPU的執行權限，程序的執行效率因此就降低了下來。

　　解決這一問題的關鍵在於，我們自己從應用程序級別檢測IO阻塞，然后切換到我們自己程序的其他任務執行，這樣把我們程序的IO降到最低，我們的程序處於就緒態就會增多，以此來迷惑操作系統，操作系統便以為我們的程序是IO比較少的程序，從而會盡可能多的分配CPU給我們，這樣也就達到了提升程序執行效率的目的。

　　實現方式：單線程+協程實現異步IO操作。

　　異步IO：就是你發起一個 網絡IO 操作，卻不用等它結束，你可以繼續做其他事情，當它結束時，你會得到通知。

 

四、 異步協程

在python3.4之后新增了asyncio模塊，可以幫我們檢測IO（只能是網絡IO【HTTP連接就是網絡IO操作】），實現應用程序級別的切換（異步IO）。注意：asyncio只能發tcp級別的請求，不能發http協議。

asyncio 是干什么的？

• 異步網絡操作
• 並發
• 協程 

 

幾個概念：

event_loop：事件循環，相當於一個無限循環，我們可以把一些函數注冊到這個事件循環上，當滿足條件發生的時候，就會調用對應的處理方法。

coroutine：中文翻譯叫協程，在 Python 中常指代為協程對象類型，我們可以將協程對象注冊到時間循環中，它會被事件循環調用。我們可以使用 async 關鍵字來定義一個方法，這個方法在調用時不會立即被執行，而是返回一個協程對象。

task：任務，它是對協程對象的進一步封裝，包含了任務的各個狀態。

future：代表將來執行或沒有執行的任務的結果，實際上和 task 沒有本質區別。

async關鍵字：async 定義一個協程；

await 關鍵字：用來掛起阻塞方法的執行。

 

注意事項：在特殊函數內部不可以出現不支持異步模塊相關的代碼。(例：time，request)

 

1.定義一個協程

import asyncio

async def execute(x):
    print('Number:', x)


coroutine = execute(1)
print('Coroutine:', coroutine)
print('After calling execute')

loop = asyncio.get_event_loop()
loop.run_until_complete(coroutine) print('After calling loop') 

運行結果：

Coroutine: <coroutine object execute at 0x1034cf830>

After calling execute

Number: 1

After calling loop

可見，async 定義的方法就會變成一個無法直接執行的 coroutine 對象，必須將其注冊到事件循環中才可以執行。

上文我們還提到了 task，它是對 coroutine 對象的進一步封裝，它里面相比 coroutine 對象多了運行狀態，比如 running、finished 等，我們可以用這些狀態來獲取協程對象的執行情況。

 

在上面的例子中，當我們將 coroutine 對象傳遞給 run_until_complete() 方法的時候，實際上它進行了一個操作就是將 coroutine 封裝成了 task 對象，我們也可以顯式地進行聲明，如下所示：

import asyncio

async def execute(x):
    print('Number:',x)
    return x

coroutine = execute(1)
print('Coroutine:',coroutine)print('After calling execute')

loop = asyncio.get_event_loop()
task = loop.create_task(coroutine) print('Task:',task)

loop.run_until_complete(task)
print('Task:',task)
print('After calling loop')

運行結果：

Coroutine: <coroutine object execute at 0x10e0f7830>

After calling execute

Task: <Task pending coro=<execute() running at demo.py:4>>

Number: 1

Task: <Task finished coro=<execute() done, defined at demo.py:4> result=1>

After calling loop

這里我們定義了 loop 對象之后，接着調用了它的 create_task() 方法將 coroutine 對象轉化為了 task 對象，隨后我們打印輸出一下，發現它是 pending 狀態。接着我們將 task 對象添加到事件循環中得到執行，隨后我們再打印輸出一下 task 對象，發現它的狀態就變成了 finished，同時還可以看到其 result 變成了 1，也就是我們定義的 execute() 方法的返回結果。

 

另外，定義 task 對象還有一種方式，就是直接通過 asyncio 的 ensure_future() 方法，返回結果也是 task 對象，這樣的話我們就可以不借助於 loop 來定義，即使我們還沒有聲明 loop 也可以提前定義好 task 對象，寫法如下：

import asyncio

async def execute(x):
    print('Number:',x)
    return x

coroutine = execute(1)
print('Coroutine:',coroutine)
print('After calling execute')

task=asyncio.ensure_future(coroutine) print('Task:',task)

loop=asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:',task)
print('After calling loop')

2.綁定回調：也可以為某個 task 綁定一個回調方法.

import asyncio
import requests

async def request():
    url='https://www.baidu.com'
    status = requests.get(url).status_code
    return status

def  callback(task):
    print('Status:',task.result())

coroutine = request()
task = asyncio.ensure_future(coroutine)
task.add_done_callback(callback) print('Task:',task)

loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:',task)

 運行結果：

Task: <Task pending coro=<request() running at demo.py:5> cb=[callback() at demo.py:11]>

Status: <Response [200]>

Task: <Task finished coro=<request() done, defined at demo.py:5> result=<Response [200]>>

在這里我們定義了一個 request() 方法，請求了百度，返回狀態碼，但是這個方法里面我們沒有任何 print() 語句。隨后我們定義了一個 callback() 方法，這個方法接收一個參數，是 task 對象，然后調用 print() 方法打印了 task 對象的結果。這樣我們就定義好了一個 coroutine 對象和一個回調方法，我們現在希望的效果是，當 coroutine 對象執行完畢之后，就去執行聲明的 callback() 方法。

那么它們二者怎樣關聯起來呢？很簡單，只需要調用 add_done_callback() 方法即可，我們將 callback() 方法傳遞給了封裝好的 task 對象，這樣當 task 執行完畢之后就可以調用 callback() 方法了，同時 task 對象還會作為參數傳遞給 callback() 方法，調用 task 對象的 result() 方法就可以獲取返回結果了。

實際上不用回調方法，直接在 task 運行完畢之后也可以直接調用 result() 方法獲取結果，運行結果是一樣的。如下所示：

import asyncio
import requests

async def request():
    url='https://www.baidu.com'
    status=requests.get(url).status_code
    return status

coroutine=request()
task=asyncio.ensure_future(coroutine)
print('Task:',task)

loop=asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task:',task)
print('Task Result:',task.result())

 

3.多任務協程

　　上面的例子我們只執行了一次請求，如果我們想執行多次請求應該怎么辦呢？我們可以定義一個 task 列表，然后使用 asyncio 的 wait() 方法即可執行。

import asyncio
import requests

async def request():
    url = 'https://www.baidu.com'
    status = requests.get(url).status_code
    return status

tasks = [asyncio.ensure_future(request()) for _ in range(5)] print('Tasks:',tasks)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
for task in tasks:
    print('Task Result:',task.result())

運行結果:

Tasks: [<Task pending coro=<request() running at demo.py:5>>, <Task pending coro=<request() running at demo.py:5>>, <Task pending coro=<request() running at demo.py:5>>, <Task pending coro=<request() running at demo.py:5>>, <Task pending coro=<request() running at demo.py:5>>]

Task Result: <Response [200]>

Task Result: <Response [200]>

Task Result: <Response [200]>

Task Result: <Response [200]>

Task Result: <Response [200]>

這里我們使用一個 for 循環創建了五個 task，組成了一個列表，然后把這個列表首先傳遞給了 asyncio 的 wait() 方法，然后再將其注冊到時間循環中，就可以發起五個任務了。

 

4.協程實現

　　上面的案例只是為后面的使用作鋪墊，接下來我們正式來看下協程在解決 IO 密集型任務上有怎樣的優勢吧！

　　為了表現出協程的優勢，我們需要先創建一個合適的實驗環境，最好的方法就是模擬一個需要等待一定時間才可以獲取返回結果的網頁，上面的代碼中使用了百度，但百度的響應太快了，而且響應速度也會受本機網速影響，所以最好的方式是自己在本地模擬一個慢速服務器，這里我們選用 Flask。

服務器代碼：
from flask import Flask
import time
 
app = Flask(__name__)
 
@app.route('/')
def index():
    time.sleep(3)
return 'Hello!'

if __name__ == '__main__':
    app.run(threaded=True)                #這表明 Flask 啟動了多線程模式，不然默認是只有一個線程的。

接下來我們再重新使用上面的方法請求一遍：

import asyncio
import requests
import time
 
start = time.time()
 
async def request():
    url = 'http://127.0.0.1:5000'
    print('Waiting for', url)
    response = requests.get(url)
    print('Get response from', url, 'Result:', response.text)
 
tasks = [asyncio.ensure_future(request()) for _ in range(5)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
 
end = time.time()
print('Cost time:', end - start)

運行結果如下：
Waiting for http://127.0.0.1:5000
Get response from http://127.0.0.1:5000 Result: Hello!
Waiting for http://127.0.0.1:5000
Get response from http://127.0.0.1:5000 Result: Hello!
Waiting for http://127.0.0.1:5000
Get response from http://127.0.0.1:5000 Result: Hello!
Waiting for http://127.0.0.1:5000
Get response from http://127.0.0.1:5000 Result: Hello!
Waiting for http://127.0.0.1:5000
Get response from http://127.0.0.1:5000 Result: Hello!
Cost time: 15.049368143081665

在這里我們還是創建了五個 task，然后將 task 列表傳給 wait() 方法並注冊到時間循環中執行。

其實，要實現異步處理，我們得先要有掛起的操作，當一個任務需要等待 IO 結果的時候，可以掛起當前任務，轉而去執行其他任務，這樣我們才能充分利用好資源，上面方法都是一本正經的串行走下來，連個掛起都沒有，怎么可能實現異步？

要實現異步，接下來我們再了解一下 await 的用法，使用 await 可以將耗時等待的操作掛起，讓出控制權。當協程執行的時候遇到 await，時間循環就會將本協程掛起，轉而去執行別的協程，直到其他的協程掛起或執行完畢。

所以，我們可能會將代碼中的 request() 方法改成如下的樣子：

async def request():
    url = 'http://127.0.0.1:5000'
    print('Waiting for', url)
    response = await requests.get(url)
print('Get response from', url, 'Result:', response.text)

僅僅是在 requests 前面加了一個 await，然而執行以下代碼，會得到如下報錯：

Waiting for http://127.0.0.1:5000 Waiting for http://127.0.0.1:5000 Waiting for http://127.0.0.1:5000 Waiting for http://127.0.0.1:5000 Waiting for http://127.0.0.1:5000 Cost time: 15.048935890197754 Task exception was never retrieved future: <Task finished coro=<request() done, defined at demo.py:7> exception=TypeError("object Response can't be used in 'await' expression",)> Traceback (most recent call last):   File "demo.py", line 10, in request     status = await requests.get(url) TypeError: object Response can't be used in 'await' expression

這次它遇到 await 方法確實掛起了，也等待了，但是最后卻報了這么個錯，這個錯誤的意思是 requests 返回的 Response 對象不能和 await 一起使用，為什么呢？因為根據官方文檔說明，await 后面的對象必須是如下格式之一：

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

reqeusts 返回的 Response 不符合上面任一條件，因此就會報上面的錯誤了。既然 await 后面可以跟一個 coroutine 對象，那么我將請求頁面的方法獨立出來，並用 async 修飾，這樣就得到了一個 coroutine 對象

import asyncio
import requests
import time
 
start = time.time()
 
async def get(url):
    return requests.get(url)
 
async def request():
    url = 'http://127.0.0.1:5000'
    print('Waiting for', url)
    response = await get(url)
    print('Get response from', url, 'Result:', response.text)
 
tasks = [asyncio.ensure_future(request()) for _ in range(5)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))
 
end = time.time()
print('Cost time:', end - start)

這里我們，我們運行一下看看：

Waiting for http://127.0.0.1:5000
Get response from http://127.0.0.1:5000 Result: Hello!
Waiting for http://127.0.0.1:5000
Get response from http://127.0.0.1:5000 Result: Hello!
Waiting for http://127.0.0.1:5000
Get response from http://127.0.0.1:5000 Result: Hello!
Waiting for http://127.0.0.1:5000
Get response from http://127.0.0.1:5000 Result: Hello!
Waiting for http://127.0.0.1:5000
Get response from http://127.0.0.1:5000 Result: Hello!
Cost time: 15.134317874908447

還是不行，它還不是異步執行，也就是說我們僅僅將涉及 IO 操作的代碼封裝到 async 修飾的方法里面是不可行的！我們必須要使用支持異步操作的請求方式才可以實現真正的異步，所以這里就需要 aiohttp 派上用場了。(由於requests 模塊不支持異步，所以用aiohttp 模塊)

 

5.使用 aiohttp 

-環境安裝：pip install aiohttp

我們將 aiohttp 用上來，將請求庫由 requests 改成了 aiohttp，通過 aiohttp 的 ClientSession 類的 get() 方法進行請求

import asyncio
import aiohttp
import time

start= time.time()

async def get(url):
    session = aiohttp.ClientSession()
    response = await session.get(url)
    result = await response.text()
    session.close()
    return result

async def request():
    url = 'http://127.0.0.1:5000'
    print('Waiting for',url)
    result = await get(url)
    print('Get response from',url,'Result:',result)

tasks = [asyncio.ensure_future(request()) for _ in range(5)]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

end = time.time()
print('Cost time:', end - start)

結果如下：我們發現這次請求的耗時由 15 秒變成了 3 秒，耗時直接變成了原來的 1/5

Waiting for http://127.0.0.1:5000 Waiting for http://127.0.0.1:5000 Waiting for http://127.0.0.1:5000 Waiting for http://127.0.0.1:5000 Waiting for http://127.0.0.1:5000 Get response from http://127.0.0.1:5000 Result: Hello! Get response from http://127.0.0.1:5000 Result: Hello! Get response from http://127.0.0.1:5000 Result: Hello! Get response from http://127.0.0.1:5000 Result: Hello! Get response from http://127.0.0.1:5000 Result: Hello! Cost time: 3.0199508666992188

代碼里面我們使用了 await，后面跟了 get() 方法，在執行這五個協程的時候，如果遇到了 await，那么就會將當前協程掛起，轉而去執行其他的協程，直到其他的協程也掛起或執行完畢，再進行下一個協程的執行。充分利用 CPU 時間，而不必把時間浪費在等待 IO 上

開始運行時，時間循環會運行第一個 task，針對第一個 task 來說，當執行到第一個 await 跟着的 get() 方法時，它被掛起，但這個 get() 方法第一步的執行是非阻塞的，掛起之后立馬被喚醒，所以立即又進入執行，創建了 ClientSession 對象，接着遇到了第二個 await，調用了 session.get() 請求方法，然后就被掛起了，由於請求需要耗時很久，所以一直沒有被喚醒，好第一個 task 被掛起了，那接下來該怎么辦呢？事件循環會尋找當前未被掛起的協程繼續執行，於是就轉而執行第二個 task 了，也是一樣的流程操作，直到執行了第五個 task 的 session.get() 方法之后，全部的 task 都被掛起了。所有 task 都已經處於掛起狀態，那咋辦？只好等待了。3 秒之后，幾個請求幾乎同時都有了響應，然后幾個 task 也被喚醒接着執行，輸出請求結果，最后耗時，3 秒！

在上面的例子中，在發出網絡請求后，既然接下來的 3 秒都是在等待的，在 3 秒之內，CPU 可以處理的 task 數量遠不止這些，那么豈不是我們放 很多 個 task 一起執行，最后得到所有結果的耗時不都是 3 秒左右嗎？因為這幾個任務被掛起后都是一起等待的。理論來說確實是這樣的，不過有個前提，那就是服務器在同一時刻接受無限次請求都能保證正常返回結果，也就是服務器無限抗壓，另外還要忽略 IO 傳輸時延，確實可以做到無限 task 一起執行且在預想時間內得到結果。

我們這里將 task 數量設置成 100，再試一下：

tasks = [asyncio.ensure_future(request()) for _ in range(100)]

耗時結果如下：
Cost time: 3.106252670288086

最后運行時間也是在 3 秒左右，當然多出來的時間就是 IO 時延了。可見，使用了異步協程之后，我們幾乎可以在相同的時間內實現成百上千倍次的網絡請求，把這個運用在爬蟲中，速度提升可謂是非常可觀了。

 

6. 與單進程、多進程對比

單進程

import requests
import time

start = time.time()


def request():
    url = 'http://127.0.0.1:5000'
    print('Waiting for', url)
    result = requests.get(url).text
    print('Get response from', url, 'Result:', result)


for _ in range(100):
    request()

end = time.time()
print('Cost time:', end - start)

最后耗時：
Cost time: 305.16639709472656

多進程

import requests
import time
import multiprocessing

start = time.time()


def request(_):
    url = 'http://127.0.0.1:5000'
    print('Waiting for', url)
    result = requests.get(url).text
    print('Get response from', url, 'Result:', result)


cpu_count = multiprocessing.cpu_count()
print('Cpu count:', cpu_count)
pool = multiprocessing.Pool(cpu_count)
pool.map(request, range(100))

end = time.time()
print('Cost time:', end - start)

這里我使用了multiprocessing 里面的 Pool 類，即進程池。我的電腦的 CPU 個數是 8 個，這里的進程池的大小就是 8。

耗時：
Cost time: 48.17306900024414

7.與多進程結合

在最新的 PyCon 2018 上，來自 Facebook 的 John Reese 介紹了 asyncio 和 multiprocessing 各自的特點，並開發了一個新的庫，叫做 aiomultiprocess。需要 Python 3.6 及更高版本才可使用。

安裝：pip install aiomultiprocess

使用這個庫，我們可以將上面的例子改寫如下：

import asyncio
import aiohttp
import time
from aiomultiprocess import Pool
 
start = time.time()
 
async def get(url):
    session = aiohttp.ClientSession()
    response = await session.get(url)
    result = await response.text()
    session.close()
    return result
 
async def request():
    url = 'http://127.0.0.1:5000'
    urls = [url for _ in range(100)]
    async with Pool() as pool:
        result = await pool.map(get, urls) return result
 
coroutine = request()
task = asyncio.ensure_future(coroutine)
loop = asyncio.get_event_loop()
loop.run_until_complete(task)
 
end = time.time()
print('Cost time:', end - start)

這樣就會同時使用多進程和異步協程進行請求，當然最后的結果其實和異步是差不多的：

Cost time: 3.1156570434570312

因為我的測試接口的原因，最快的響應也是 3 秒，所以這部分多余的時間基本都是 IO 傳輸時延。但在真實情況下，我們在做爬取的時候遇到的情況千變萬化，一方面我們使用異步協程來防止阻塞，另一方面我們使用 multiprocessing 來利用多核成倍加速，節省時間其實還是非常可觀的。

 

更多案例

import aiohttp
import asyncio
from lxml import etree

all_titles = []

headers = {
    'User-Agent':'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/70.0.3538.77 Safari/537.36'

}
async def request(url):
    async with aiohttp.ClientSession() as s:
        async with await s.get(url,headers=headers) as response:
            page_text = await response.text()
            return page_text


def parse(task):
    page_text = task.result()
    page_text = page_text.encode('gb2312').decode('gbk')
    tree = etree.HTML(page_text)
    tr_list = tree.xpath('//*[@id="morelist"]/div/table[2]//tr/td/table//tr')
    for tr in tr_list:
        title = tr.xpath('./td[2]/a[2]/text()')[0]
        print(title)
        all_titles.append(title)

urls = []
url = 'http://wz.sun0769.com/index.php/question/questionType?type=4&page=%d'
for page in range(100):
    u_page = page * 30
    new_url = format(url%u_page)
    urls.append(new_url)

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

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

 

 

參考鏈接： https://blog.csdn.net/zhusongziye/article/details/81637088