Python 異步網絡爬蟲 I

本文主要討論下面幾個問題：

• 什麼是異步（Asynchronous）編程？
• 爲何要使用異步編程？
• 在 Python 中有哪些實現異步編程的方法？
• Python 3.5 如何使用 async/await 實現異步網絡爬蟲？

所謂異步是相對於同步（Synchronous）的概念來講的，之因此容易形成混亂，是由於剛開始接觸這兩個概念時容易把同步看作是同時，而同時不是意味着並行（Parallel）嗎？然而實際上同步或者異步是針對於時間軸的概念，同步意味着順序、統一的時間軸，而異步則意味着亂序、效率優先的時間軸。好比在爬蟲運行時，先抓取 A 頁面，而後從中提取下一層頁面 B 的連接，此時的爬蟲程序的運行只能是同步的，B 頁面只能等到 A 頁面處理完成以後才能抓取；然而對於獨立的兩個頁面 A1 和 A2，在處理 A1 網絡請求的時間裏，與其讓 CPU 空閒而 A2 等在後面，不如先處理 A2，等到誰先完成網絡請求誰就先來進行處理，這樣能夠更加充分地利用 CPU，可是 A1 和 A2 的執行順序則是不肯定的，也就是異步的。

很顯然，在某些狀況下采用異步編程能夠提升程序運行效率，減小沒必要要的等待時間，而之因此可以作到這一點，是由於計算機的 CPU 與其它設備是獨立運做的，同時 CPU 的運行效率遠高於其餘設備的讀寫（I/O）效率。爲了利用異步編程的優點，人們想出了不少方法來從新安排、調度（Schedule）程序的運行順序，從而最大化 CPU 的使用率，其中包括進程、線程、協程等（具體可參考《Python 中的進程、線程、協程、同步、異步、回調》）。在 Python 3.5 之前經過 @types.coroutine 做爲修飾器的方式將一個生成器（Generator）轉化爲一個協程，而在 Python 3.5 中則經過關鍵詞 async/await 來定義一個協程，同時也將 asyncio 歸入爲標準庫，用於實現基於協程的異步編程。

要使用 asyncio 須要理解下面幾個概念：

• Event loop
• Coroutine
• Future & Task

Event loop

瞭解 JavaScript 或 Node.js 確定對事件循環不陌生，咱們能夠把它看做是一種循環式（loop）的調度機制，它能夠安排鬚要 CPU 執行的操做優先執行，而會被 I/O 阻塞的行爲則進入等待隊列：

asyncio 自帶了事件循環：

import asyncio

loop = anscio.get_event_loop()
# loop.run_until_complete(coro())
loop.close()複製代碼

固然你也能夠選擇其它的實現形式，例如 Sanic 框架採用的 uvloop，用起來也很是簡單（ 至於性能上是否更優我沒有驗證過，但至少在 Jupyter Notebook 上 uvloop 用起來更方便）：

import asyncio
import uvloop

loop = uvloop.new_event_loop()
asyncio.set_event_loop(loop)複製代碼

Coroutine

Python 3.5 之後推薦使用 async/await 關鍵詞來定義協程，它具備以下特性：

• 經過 await 將可能阻塞的行爲掛起，直到有結果以後繼續執行，Event loop 也是據此來對多個協程的執行進行調度的；
• 協程並不像通常的函數同樣，經過 coro() 進行調用並不會執行它，而只有將它放入 Event loop 進行調度才能執行。

一個簡單的例子：

import uvloop
import asyncio

loop = uvloop.new_event_loop()
asyncio.set_event_loop(loop)

async def compute(a, b):
    print("Computing {} + {}...".format(a, b))
    await asyncio.sleep(a+b)
    return a + b
tasks = []
for i, j in zip(range(3), range(3)):
    print(i, j)
    tasks.append(compute(i, j))
loop.run_until_complete(asyncio.gather(*tasks))
loop.close()

### OUTPUT
""" 0 0 1 1 2 2 Computing 0 + 0... Computing 1 + 1... Computing 2 + 2... CPU times: user 1.05 ms, sys: 1.21 ms, total: 2.26 ms Wall time: 4 s """複製代碼

因爲咱們沒辦法知道協程將在何時調用及返回，asyncio 中提供了 Future 這一對象來追蹤它的執行結果。

Future & Task

Future 至關於 JavaScript 中的 Promise，用於保存將來可能返回的結果。而 Task 則是 Future 的子類，與 Future 不一樣的是它包含了一個將要執行的協程（ 從而組成一個須要被調度的任務）。還以上面的程序爲例，若是想要知道計算結果，能夠經過 asyncio.ensure_future() 方法將協程包裹成 Task，最後再來讀取結果：

import uvloop
import asyncio

loop = uvloop.new_event_loop()
asyncio.set_event_loop(loop)

async def compute(a, b):
    print("Computing {} + {}...".format(a, b))
    await asyncio.sleep(a+b)
    return a + b
tasks = []
for i, j in zip(range(3), range(3)):
    print(i, j)
    tasks.append(asyncio.ensure_future(compute(i, j)))
loop.run_until_complete(asyncio.gather(*tasks))
for t in tasks:
    print(t.result())
loop.close()

### OUTPUT
""" 0 0 1 1 2 2 Computing 0 + 0... Computing 1 + 1... Computing 2 + 2... 0 2 4 CPU times: user 1.62 ms, sys: 1.86 ms, total: 3.49 ms Wall time: 4.01 s """複製代碼

異步網絡請求

Python 處理網絡請求最好用的庫就是 requests（應該沒有之一），但因爲它的請求過程是同步阻塞的，所以只好選用 aiohttp。爲了對比同步與異步狀況下的差別，先僞造一個假的異步處理服務器：

from sanic import Sanic
from sanic.response import text
import asyncio
app = Sanic(__name__)

@app.route("/<word>")
@app.route("/")
async def index(req, word=""):
    t = len(word) / 10
    await asyncio.sleep(t)
    return text("It costs {}s to process `{}`!".format(t, word))
app.run()複製代碼

服務器處理耗時與請求參數（word）長度成正比，採用同步請求方式，運行結果以下：

import requests as req

URL = "http://127.0.0.1:8000/{}"
words = ["Hello", "Python", "Fans", "!"]

for word in words:
    resp = req.get(URL.format(word))
    print(resp.text)

### OUTPUT
""" It costs 0.5s to process `Hello`! It costs 0.6s to process `Python`! It costs 0.4s to process `Fans`! It costs 0.1s to process `!`! CPU times: user 18.5 ms, sys: 2.98 ms, total: 21.4 ms Wall time: 1.64 s """複製代碼

採用異步請求，運行結果以下：

import asyncio
import aiohttp
import uvloop

URL = "http://127.0.0.1:8000/{}"
words = ["Hello", "Python", "Fans", "!"]

async def getPage(session, word):
    with aiohttp.Timeout(10):
        async with session.get(URL.format(word)) as resp:
            print(await resp.text())

loop = uvloop.new_event_loop()
asyncio.set_event_loop(loop)
session = aiohttp.ClientSession(loop=loop)

tasks = []
for word in words:
    tasks.append(getPage(session, word))

loop.run_until_complete(asyncio.gather(*tasks))

loop.close()
session.close()

### OUTPUT
""" It costs 0.1s to process `!`! It costs 0.4s to process `Fans`! It costs 0.5s to process `Hello`! It costs 0.6s to process `Python`! CPU times: user 61.2 ms, sys: 18.2 ms, total: 79.3 ms Wall time: 732 ms """複製代碼

從運行時間上來看效果是很明顯的。

未完待續

接下來將對 aiohttp 進行簡單封裝，更有利於假裝成普通瀏覽器用戶訪問，從而服務於爬蟲發送網絡請求。

【閱讀原文】

參考

1. Python 中的進程、線程、協程、同步、異步、回調
2. Asyncio Document
3. aiohttp - HTTP Client