轉載自 崔慶才的個人站點-Python 中非同步協程的使用方法介紹

1. 前言

在執行一些 IO 密集型任務的時候，程式常常會因為等待 IO 而阻塞。比如在網路爬蟲中，如果我們使用 requests 庫來進行請求的話，如果網站響應速度過慢，程式一直在等待網站響應，最後導致其爬取效率是非常非常低的。 為了解決這類問題，本文就來探討一下 Python 中非同步協程來加速的方法，此種方法對於 IO 密集型任務非常有效。如將其應用到網路爬蟲中，爬取效率甚至可以成百倍地提升。 注：本文協程使用 async/await 來實現，需要 Python 3.5 及以上版本。

2. 基本概念

在瞭解非同步協程之前，我們首先得了解一些基礎概念，如阻塞和非阻塞、同步和非同步、多程序和協程。

2.1 阻塞

阻塞狀態指程式未得到所需計算資源時被掛起的狀態。程式在等待某個操作完成期間，自身無法繼續幹別的事情，則稱該程式在該操作上是阻塞的。 常見的阻塞形式有：網路 I/O 阻塞、磁碟 I/O 阻塞、使用者輸入阻塞等。阻塞是無處不在的，包括 CPU 切換上下文時，所有的程序都無法真正幹事情，它們也會被阻塞。如果是多核 CPU 則正在執行上下文切換操作的核不可被利用。

2.2 非阻塞

程式在等待某操作過程中，自身不被阻塞，可以繼續執行幹別的事情，則稱該程式在該操作上是非阻塞的。 非阻塞並不是在任何程式級別、任何情況下都可以存在的。 僅當程式封裝的級別可以囊括獨立的子程式單元時，它才可能存在非阻塞狀態。 非阻塞的存在是因為阻塞存在，正因為某個操作阻塞導致的耗時與效率低下，我們才要把它變成非阻塞的。

2.3 同步

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

2.4 非同步

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

2.5 多程序

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

2.6 協程

協程，英文叫做 Coroutine，又稱微執行緒，纖程，協程是一種使用者態的輕量級執行緒。 協程擁有自己的暫存器上下文和棧。協程排程切換時，將暫存器上下文和棧儲存到其他地方，在切回來的時候，恢復先前儲存的暫存器上下文和棧。因此協程能保留上一次呼叫時的狀態，即所有區域性狀態的一個特定組合，每次過程重入時，就相當於進入上一次呼叫的狀態。 協程本質上是個單程序，協程相對於多程序來說，無需執行緒上下文切換的開銷，無需原子操作鎖定及同步的開銷，程式設計模型也非常簡單。 我們可以使用協程來實現非同步操作，比如在網路爬蟲場景下，我們發出一個請求之後，需要等待一定的時間才能得到響應，但其實在這個等待過程中，程式可以幹許多其他的事情，等到響應得到之後才切換回來繼續處理，這樣可以充分利用 CPU 和其他資源，這就是非同步協程的優勢。

3. 非同步協程用法

接下來讓我們來了解下協程的實現，從 Python 3.4 開始，Python 中加入了協程的概念，但這個版本的協程還是以生成器物件為基礎的，在 Python 3.5 則增加了 async/await，使得協程的實現更加方便。 Python 中使用協程最常用的庫莫過於 asyncio，所以本文會以 asyncio 為基礎來介紹協程的使用。 首先我們需要了解下面幾個概念：

• event_loop：事件迴圈，相當於一個無限迴圈，我們可以把一些函式註冊到這個事件迴圈上，當滿足條件發生的時候，就會呼叫對應的處理方法。
• coroutine：中文翻譯叫協程，在 Python 中常指代為協程物件型別，我們可以將協程物件註冊到時間迴圈中，它會被事件迴圈呼叫。我們可以使用 async 關鍵字來定義一個方法，這個方法在呼叫時不會立即被執行，而是返回一個協程物件。
• task：任務，它是對協程物件的進一步封裝，包含了任務的各個狀態。
• future：代表將來執行或沒有執行的任務的結果，實際上和 task 沒有本質區別。
  另外我們還需要了解 async/await 關鍵字，它是從 Python 3.5 才出現的，專門用於定義協程。其中，async 定義一個協程，await 用來掛起阻塞方法的執行。

3.1 定義協程

首先我們來定義一個協程，體驗一下它和普通程序在實現上的不同之處，程式碼如下：

import asyncio

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

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

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

執行結果：

Coroutine: <coroutine object execute at 0x7ff660045e40>
Number: 1
After loop

首先我們引入了 asyncio 這個包，這樣我們才可以使用 async 和 await，然後我們使用 async 定義了一個 execute () 方法，方法接收一個數字引數，方法執行之後會列印這個數字。 隨後我們直接呼叫了這個方法，然而這個方法並沒有執行，而是返回了一個 coroutine 協程物件。隨後我們使用 get_event_loop () 方法建立了一個事件迴圈 loop，並呼叫了 loop 物件的 run_until_complete () 方法將協程註冊到事件迴圈 loop 中，然後啟動。最後我們才看到了 execute () 方法列印了輸出結果。 可見，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)

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

Output:

Coroutine: <coroutine object execute at 0x7f88a0053040>
Task: <Task pending name='Task-1' coro=<execute() running at hello_world.py:3>>
Number: 1
Task after: <Task finished name='Task-1' coro=<execute() done, defined at hello_world.py:3> result=1>
After 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)

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

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

Output:

Coroutine: <coroutine object execute at 0x7fb0b0263040>
Task: <Task pending name='Task-1' coro=<execute() running at hello_world.py:3>>
Number: 1
Task after: <Task finished name='Task-1' coro=<execute() done, defined at hello_world.py:3> result=1>
After loop

發現其效果都是一樣的。

3.2 繫結回撥

另外我們也可以為某個 task 繫結一個回撥方法，來看下面的例子：

import asyncio
from asyncio import tasks
import requests

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

def callback(task):
    print(task.result())

coroutine = request()
task = asyncio.ensure_future(coroutine)
task.add_done_callback(callback)  # 給task添加回調函式
print('Task:', task)

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

在這裡我們定義了一個 request () 方法，請求了百度，返回狀態碼，但是這個方法裡面我們沒有任何 print () 語句。隨後我們定義了一個 callback () 方法，這個方法接收一個引數，是 task 物件，然後呼叫 print () 方法列印了 task 物件的結果。這樣我們就定義好了一個 coroutine 物件和一個回撥方法，我們現在希望的效果是，當 coroutine 物件執行完畢之後，就去執行宣告的 callback () 方法。 那麼它們二者怎樣關聯起來呢？很簡單，只需要呼叫 add_done_callback () 方法即可，我們將 callback () 方法傳遞給了封裝好的 task 物件，這樣當 task 執行完畢之後就可以呼叫 callback () 方法了，同時 task 物件還會作為引數傳遞給 callback () 方法，呼叫 task 物件的 result () 方法就可以獲取返回結果了。 執行結果：
Output:

Task: <Task pending name='Task-1' coro=<request() running at callback.py:5> cb=[callback() at callback.py:10]>
<Response [200]>
Task after: <Task finished name='Task-1' coro=<request() done, defined at callback.py:5> result=<Response [200]>>
After loop

實際上不用回撥方法，直接在 task 執行完畢之後也可以直接呼叫 result () 方法獲取結果，如下所示：

import asyncio
from asyncio import tasks
import requests

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

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

loop = asyncio.get_event_loop()
loop.run_until_complete(task)
print('Task after:', task)
print('Task result:', task.result())   # 也可以直接獲取結果

執行結果是一樣的.

3.3 多工協程

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

import asyncio
from asyncio import tasks
import requests

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

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

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

for task in tasks:
    print(task.result())

print('Done')

這裡我們使用一個 for 迴圈建立了五個 task，組成了一個列表，然後把這個列表首先傳遞給了 asyncio 的 wait () 方法，然後再將其註冊到時間迴圈中，就可以發起五個任務了。最後我們再將任務的執行結果輸出出來，執行結果如下：

Tasks: [<Task pending name='Task-1' coro=<request() running at callback.py:5>>, <Task pending name='Task-2' coro=<request() running at callback.py:5>>, <Task pending name='Task-3' coro=<request() running at callback.py:5>>, <Task pending name='Task-4' coro=<request() running at callback.py:5>>, <Task pending name='Task-5' coro=<request() running at callback.py:5>>, <Task pending name='Task-6' coro=<request() running at callback.py:5>>, <Task pending name='Task-7' coro=<request() running at callback.py:5>>, <Task pending name='Task-8' coro=<request() running at callback.py:5>>, <Task pending name='Task-9' coro=<request() running at callback.py:5>>, <Task pending name='Task-10' coro=<request() running at callback.py:5>>]
<Response [200]>
<Response [200]>
<Response [200]>
<Response [200]>
<Response [200]>
<Response [200]>
<Response [200]>
<Response [200]>
<Response [200]>
<Response [200]>
Done

可以看到五個任務被順次執行了，並得到了執行結果。

3.4 協程實現

前面說了這麼一通，又是 async，又是 coroutine，又是 task，又是 callback，但似乎並沒有看出協程的優勢啊？反而寫法上更加奇怪和麻煩了，別急，上面的案例只是為後面的使用作鋪墊，接下來我們正式來看下協程在解決 IO 密集型任務上有怎樣的優勢吧！ 上面的程式碼中，我們用一個網路請求作為示例，這就是一個耗時等待的操作，因為我們請求網頁之後需要等待頁面響應並返回結果。耗時等待的操作一般都是 IO 操作，比如檔案讀取、網路請求等等。協程對於處理這種操作是有很大優勢的，當遇到需要等待的情況的時候，程式可以暫時掛起，轉而去執行其他的操作，從而避免一直等待一個程式而耗費過多的時間，充分利用資源。 為了表現出協程的優勢，我們需要先建立一個合適的實驗環境，最好的方法就是模擬一個需要等待一定時間才可以獲取返回結果的網頁，上面的程式碼中使用了百度，但百度的響應太快了，而且響應速度也會受本機網速影響，所以最好的方式是自己在本地模擬一個慢速伺服器，這裡我們選用 Flask。 如果沒有安裝 Flask 的話可以執行如下命令安裝：

pip3 install flask

然後編寫伺服器程式碼如下：

from flask import Flask
import time

app = Flask(__name__)

@app.route('/')
def index():
    time.sleep(3)
    return '<h1>Hello World!</h1>'

if __name__ == '__main__':
    app.run(threaded=True)

這裡我們定義了一個 Flask 服務，主入口是 index () 方法，方法裡面先呼叫了 sleep () 方法休眠 3 秒，然後接著再返回結果，也就是說，每次請求這個介面至少要耗時 3 秒，這樣我們就模擬了一個慢速的服務介面。 注意這裡服務啟動的時候，run () 方法加了一個引數 threaded，這表明 Flask 啟動了多執行緒模式，不然預設是隻有一個執行緒的。如果不開啟多執行緒模式，同一時刻遇到多個請求的時候，只能順次處理，這樣即使我們使用協程非同步請求了這個服務，也只能一個一個排隊等待，瓶頸就會出現在服務端。所以，多執行緒模式是有必要開啟的。 啟動之後，Flask 應該預設會在 127.0.0.1:5000 上執行，執行之後控制檯輸出結果如下：

*Runningonhttp://127.0.0.1:5000/(PressCTRL+Ctoquit)

import asyncio
import requests
import time 

start = time.time()

async def request():
    url = 'http://127.0.0.1:5000'
    print("wait for", url)
    r = requests.get(url)
    print("response", r.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)

在這裡我們還是建立了五個 task，然後將 task 列表傳給 wait () 方法並註冊到時間迴圈中執行。 執行結果如下：

wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
Cost time: 15.068505048751831

可以發現和正常的請求並沒有什麼兩樣，依然還是順次執行的，耗時 15 秒，平均一個請求耗時 3 秒，說好的非同步處理呢？ 其實，要實現非同步處理，我們得先要有掛起的操作，當一個任務需要等待 IO 結果的時候，可以掛起當前任務，轉而去執行其他任務，這樣我們才能充分利用好資源，上面方法都是一本正經的序列走下來，連個掛起都沒有，怎麼可能實現非同步？想太多了。 要實現非同步，接下來我們再瞭解一下 await 的用法，使用 await 可以將耗時等待的操作掛起，讓出控制權。當協程執行的時候遇到 await，時間迴圈就會將本協程掛起，轉而去執行別的協程，直到其他的協程掛起或執行完畢。 所以，我們可能會將程式碼中的 request () 方法改成如下的樣子：

async def request():
    url = 'http://127.0.0.1:5000'
    print("wait for", url)
    r = await requests.get(url)
    print("response", r.text)

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

wait for http://127.0.0.1:5000
wait for http://127.0.0.1:5000
wait for http://127.0.0.1:5000
wait for http://127.0.0.1:5000
wait for http://127.0.0.1:5000
Cost time: 15.081188201904297
Task exception was never retrieved
future: <Task finished name='Task-1' coro=<request() done, defined at client.py:10> exception=TypeError("object Response can't be used in 'await' expression")>
Traceback (most recent call last):
  File "client.py", line 13, in request
    r = 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 方法的物件返回的一個迭代器。
  可以參見：https://www.python.org/dev/peps/pep-0492/#await-expression。 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("wait for", url)
    r = await get(url)
    print("response", r.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)

這裡我們將請求頁面的方法獨立出來，並用 async 修飾，這樣就得到了一個 coroutine 物件，我們執行一下看看：

wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
Cost time: 15.072779178619385

還是不行，它還不是非同步執行，也就是說我們僅僅將涉及 IO 操作的程式碼封裝到 async 修飾的方法裡面是不可行的！我們必須要使用支援非同步操作的請求方式才可以實現真正的非同步，所以這裡就需要 aiohttp 派上用場了。

3.5 使用 aiohttp

aiohttp 是一個支援非同步請求的庫，利用它和 asyncio 配合我們可以非常方便地實現非同步請求操作。 安裝方式如下：

pip3 install aiohttp

官方文件連結為：https://aiohttp.readthedocs.io/，它分為兩部分，一部分是 Client，一部分是 Server，詳細的內容可以參考官方文件。 下面我們將 aiohttp 用上來，將程式碼改成如下樣子：

import asyncio
import requests
import time 
import aiohttp

start = time.time()

async def get(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def request():
    url = 'http://127.0.0.1:5000'
    print("wait for", url)
    result = await get(url)
    print("response", 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)

在這裡我們將請求庫由 requests 改成了 aiohttp，通過 aiohttp 的 ClientSession 類的 get () 方法進行請求，結果如下：

wait for http://127.0.0.1:5000
wait for http://127.0.0.1:5000
wait for http://127.0.0.1:5000
wait for http://127.0.0.1:5000
wait for http://127.0.0.1:5000
response <h1>Hello World!</h1>
response <h1>Hello World!</h1>
response <h1>Hello World!</h1>
response <h1>Hello World!</h1>
response <h1>Hello World!</h1>
Cost time: 3.020137071609497

成功了！我們發現這次請求的耗時由 15 秒變成了 3 秒，耗時直接變成了原來的 1/5。 程式碼裡面我們使用了 await，後面跟了 get () 方法，在執行這五個協程的時候，如果遇到了 await，那麼就會將當前協程掛起，轉而去執行其他的協程，直到其他的協程也掛起或執行完畢，再進行下一個協程的執行。 開始執行時，時間迴圈會執行第一個 task，針對第一個 task 來說，當執行到第一個 await 跟著的 get () 方法時，它被掛起，但這個 get () 方法第一步的執行是非阻塞的，掛起之後立馬被喚醒，所以立即又進入執行，建立了 ClientSession 物件，接著遇到了第二個 await，呼叫了 session.get () 請求方法，然後就被掛起了，由於請求需要耗時很久，所以一直沒有被喚醒，好第一個 task 被掛起了，那接下來該怎麼辦呢？事件迴圈會尋找當前未被掛起的協程繼續執行，於是就轉而執行第二個 task 了，也是一樣的流程操作，直到執行了第五個 task 的 session.get () 方法之後，全部的 task 都被掛起了。所有 task 都已經處於掛起狀態，那咋辦？只好等待了。3 秒之後，幾個請求幾乎同時都有了響應，然後幾個 task 也被喚醒接著執行，輸出請求結果，最後耗時，3 秒！ 怎麼樣？這就是非同步操作的便捷之處，當遇到阻塞式操作時，任務被掛起，程式接著去執行其他的任務，而不是傻傻地等著，這樣可以充分利用 CPU 時間，而不必把時間浪費在等待 IO 上。 有人就會說了，既然這樣的話，在上面的例子中，在發出網路請求後，既然接下來的 3 秒都是在等待的，在 3 秒之內，CPU 可以處理的 task 數量遠不止這些，那麼豈不是我們放 10 個、20 個、50 個、100 個、1000 個 task 一起執行，最後得到所有結果的耗時不都是 3 秒左右嗎？因為這幾個任務被掛起後都是一起等待的。 理論來說確實是這樣的，不過有個前提，那就是伺服器在同一時刻接受無限次請求都能保證正常返回結果，也就是伺服器無限抗壓，另外還要忽略 IO 傳輸時延，確實可以做到無限 task 一起執行且在預想時間內得到結果。 我們這裡將 task 數量設定成 100，再試一下：

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

耗時結果如下：

Cost time: 3.087064027786255

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

3.6 與單程序、多程序對比

可能有的小夥伴非常想知道上面的例子中，如果 100 次請求，不是用非同步協程的話，使用單程序和多程序會耗費多少時間，我們來測試一下： 首先來測試一下單程序的時間：

import requests
import time 

start = time.time()


def request():
    url = 'http://127.0.0.1:5000'
    print("wait for", url)
    r = requests.get(url)
    print("response", r.text)

for _ in range(100):
    request()

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

最後耗時：

Cost time: 301.2694022655487

接下來我們使用多程序來測試下，使用 multiprocessing 庫：

import requests
import time 
import multiprocessing

start = time.time()

def request(_):
    url = 'http://127.0.0.1:5000'
    print("wait for", url)
    r = requests.get(url)
    print("response", r.text)

if __name__ == '__main__':
    cpu_count = multiprocessing.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.47766995429993

我也在4核的樹莓派上測試，執行時間是 Cost time: 84.7542974948883
可見 multiprocessing 相比單執行緒來說，還是可以大大提高效率的。

3.7 與多程序的結合

既然非同步協程和多程序對網路請求都有提升，那麼為什麼不把二者結合起來呢？在最新的 PyCon 2018 上，來自 Facebook 的 John Reese 介紹了 asyncio 和 multiprocessing 各自的特點，並開發了一個新的庫，叫做 aiomultiprocess，感興趣的可以瞭解下：https://www.youtube.com/watch?v=0kXaLh8Fz3k。 這個庫的安裝方式是：

pip3 install aiomultiprocess

需要 Python 3.6 及更高版本才可使用。 使用這個庫，我們可以將上面的例子改寫如下：

import asyncio
import time 
import aiohttp
from aiomultiprocess import Pool

start = time.time()

async def get(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def request():
    url = 'http://127.0.0.1:5000'
    urls = [url for _ in range(100)]
    async with Pool() as pool:
        results = await pool.map(get, urls)
        print(results)

if __name__ == '__main__':
    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.548332929611206

因為我的測試介面的原因，最快的響應也是 3 秒，所以這部分多餘的時間基本都是 IO 傳輸時延。但在真實情況下，我們在做爬取的時候遇到的情況千變萬化，一方面我們使用非同步協程來防止阻塞，另一方面我們使用 multiprocessing 來利用多核成倍加速，節省時間其實還是非常可觀的。 以上便是 Python 中協程的基本用法，希望對大家有幫助。