Python使用Asyncio實現檢查網站狀態

2023-03-31 06:01:41

1. 如何使用 Asyncio 檢查 HTTP 狀態

asyncio 模組提供了對開啟通訊端連線和通過流讀寫資料的支援。我們可以使用此功能來檢查網頁的狀態。

這可能涉及四個步驟，它們是：

開啟一個連線
寫一個請求
讀取響應
關閉連線

2. 開啟 HTTP 連線

可以使用 asyncio.open_connection() 函數在 asyncio 中開啟連線。在眾多引數中，該函數採用字串主機名和整數埠號。

這是一個必須等待的協程，它返回一個 StreamReader 和一個 StreamWriter，用於使用通訊端進行讀寫。

這可用於在埠 80 上開啟 HTTP 連線。

...
# open a socket connection
reader, writer = await asyncio.open_connection('www.google.com', 80)

我們還可以使用 ssl=True 引數開啟 SSL 連線。這可用於在埠 443 上開啟 HTTPS 連線。

...
# open a socket connection
reader, writer = await asyncio.open_connection('www.google.com', 443)

3. 寫入 HTTP 請求

開啟後，我們可以向 StreamWriter 寫入查詢以發出 HTTP 請求。例如，HTTP 版本 1.1 請求是純文字格式的。我們可以請求檔案路徑“/”，它可能如下所示：

GET / HTTP/1.1
Host: www.google.com

重要的是，每行末尾必須有一個回車和一個換行符（rn），末尾有一個空行。

作為 Python 字串，這可能如下所示：

'GET / HTTP/1.1rn'
'Host: www.google.comrn'
'rn'

在寫入 StreamWriter 之前，此字串必須編碼為位元組。這可以通過對字串本身使用 encode() 方法來實現。預設的“utf-8”編碼可能就足夠了。

...
# encode string as bytes
byte_data = string.encode()

然後可以通過 StreamWriter 的 write() 方法將位元組寫入通訊端。

...
# write query to socket
writer.write(byte_data)

寫入請求後，最好等待位元組資料傳送完畢並等待通訊端準備就緒。這可以通過 drain() 方法來實現。這是一個必須等待的協程。

...
# wait for the socket to be ready.
await writer.drain()

4. 讀取 HTTP 響應

發出 HTTP 請求後，我們可以讀取響應。這可以通過通訊端的 StreamReader 來實現。可以使用讀取一大塊位元組的 read() 方法或讀取一行位元組的 readline() 方法來讀取響應。

我們可能更喜歡 readline() 方法，因為我們使用的是基於文字的 HTTP 協定，它一次傳送一行 HTML 資料。readline() 方法是協程，必須等待。

...
# read one line of response
line_bytes = await reader.readline()

HTTP 1.1 響應由兩部分組成，一個由空行分隔的檔頭，然後是一個空行終止的主體。header 包含有關請求是否成功以及將傳送什麼型別的檔案的資訊，body 包含檔案的內容，例如 HTML 網頁。

HTTP 檔頭的第一行包含伺服器上所請求頁面的 HTTP 狀態。每行都必須從位元組解碼為字串。

這可以通過對位元組資料使用 decode() 方法來實現。同樣，預設編碼為“utf_8”。

...
# decode bytes into a string
line_data = line_bytes.decode()

5. 關閉 HTTP 連線

我們可以通過關閉 StreamWriter 來關閉通訊端連線。這可以通過呼叫 close() 方法來實現。

...
# close the connection
writer.close()

這不會阻塞並且可能不會立即關閉通訊端。現在我們知道如何使用 asyncio 發出 HTTP 請求和讀取響應，讓我們看一些檢查網頁狀態的範例。

6. 順序檢查 HTTP 狀態的範例

我們可以開發一個範例來使用 asyncio 檢查多個網站的 HTTP 狀態。

在此範例中，我們將首先開發一個協程來檢查給定 URL 的狀態。然後我們將為排名前 10 的網站中的每一個呼叫一次這個協程。

首先，我們可以定義一個協程，它將接受一個 URL 字串並返回 HTTP 狀態。

# get the HTTP/S status of a webpage
async def get_status(url):
	# ...

必須將 URL 解析為其組成部分。我們在發出 HTTP 請求時需要主機名和檔案路徑。我們還需要知道 URL 方案（HTTP 或 HTTPS）以確定是否需要 SSL。

這可以使用 urllib.parse.urlsplit() 函數來實現，該函數接受一個 URL 字串並返回所有 URL 元素的命名元組。

...
# split the url into components
url_parsed = urlsplit(url)

然後我們可以開啟基於 URL 方案的 HTTP 連線並使用 URL 主機名。

...
# open the connection
if url_parsed.scheme == 'https':
    reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
else:
    reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)

接下來，我們可以使用主機名和檔案路徑建立 HTTP GET 請求，並使用 StreamWriter 將編碼位元組寫入通訊端。

...
# send GET request
query = f'GET {url_parsed.path} HTTP/1.1rnHost: {url_parsed.hostname}rnrn'
# write query to socket
writer.write(query.encode())
# wait for the bytes to be written to the socket
await writer.drain()

接下來，我們可以讀取 HTTP 響應。我們只需要包含 HTTP 狀態的響應的第一行。

...
# read the single line response
response = await reader.readline()

然後可以關閉連線。

...
# close the connection
writer.close()

最後，我們可以解碼從伺服器讀取的位元組、遠端尾隨空白，並返回 HTTP 狀態。

...
# decode and strip white space
status = response.decode().strip()
# return the response
return status

將它們結合在一起，下面列出了完整的 get_status() 協程。它沒有任何錯誤處理，例如無法存取主機或響應緩慢的情況。這些新增將為讀者提供一個很好的擴充套件。

# get the HTTP/S status of a webpage
async def get_status(url):
    # split the url into components
    url_parsed = urlsplit(url)
    # open the connection
    if url_parsed.scheme == 'https':
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
    else:
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)
    # send GET request
    query = f'GET {url_parsed.path} HTTP/1.1rnHost: {url_parsed.hostname}rnrn'
    # write query to socket
    writer.write(query.encode())
    # wait for the bytes to be written to the socket
    await writer.drain()
    # read the single line response
    response = await reader.readline()
    # close the connection
    writer.close()
    # decode and strip white space
    status = response.decode().strip()
    # return the response
    return status

接下來，我們可以為我們要檢查的多個網頁或網站呼叫 get_status() 協程。在這種情況下，我們將定義一個世界排名前 10 的網頁列表。

...
# list of top 10 websites to check
sites = ['https://www.google.com/',
    'https://www.youtube.com/',
    'https://www.facebook.com/',
    'https://twitter.com/',
    'https://www.instagram.com/',
    'https://www.baidu.com/',
    'https://www.wikipedia.org/',
    'https://yandex.ru/',
    'https://yahoo.com/',
    'https://www.whatsapp.com/'
    ]

然後我們可以使用我們的 get_status() 協程依次查詢每個。在這種情況下，我們將在一個迴圈中按順序這樣做，並依次報告每個狀態。

...
# check the status of all websites
for url in sites:
    # get the status for the url
    status = await get_status(url)
    # report the url and its status
    print(f'{url:30}:t{status}')

在使用 asyncio 時，我們可以做得比順序更好，但這提供了一個很好的起點，我們可以在以後進行改進。將它們結合在一起，main() 協程查詢前 10 個網站的狀態。

# main coroutine
async def main():
    # list of top 10 websites to check
    sites = ['https://www.google.com/',
        'https://www.youtube.com/',
        'https://www.facebook.com/',
        'https://twitter.com/',
        'https://www.instagram.com/',
        'https://www.baidu.com/',
        'https://www.wikipedia.org/',
        'https://yandex.ru/',
        'https://yahoo.com/',
        'https://www.whatsapp.com/'
        ]
    # check the status of all websites
    for url in sites:
        # get the status for the url
        status = await get_status(url)
        # report the url and its status
        print(f'{url:30}:t{status}')

最後，我們可以建立 main() 協程並將其用作 asyncio 程式的入口點。

...
# run the asyncio program
asyncio.run(main())

將它們結合在一起，下面列出了完整的範例。

# SuperFastPython.com
# check the status of many webpages
import asyncio
from urllib.parse import urlsplit
 
# get the HTTP/S status of a webpage
async def get_status(url):
    # split the url into components
    url_parsed = urlsplit(url)
    # open the connection
    if url_parsed.scheme == 'https':
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
    else:
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)
    # send GET request
    query = f'GET {url_parsed.path} HTTP/1.1rnHost: {url_parsed.hostname}rnrn'
    # write query to socket
    writer.write(query.encode())
    # wait for the bytes to be written to the socket
    await writer.drain()
    # read the single line response
    response = await reader.readline()
    # close the connection
    writer.close()
    # decode and strip white space
    status = response.decode().strip()
    # return the response
    return status
 
# main coroutine
async def main():
    # list of top 10 websites to check
    sites = ['https://www.google.com/',
        'https://www.youtube.com/',
        'https://www.facebook.com/',
        'https://twitter.com/',
        'https://www.instagram.com/',
        'https://www.baidu.com/',
        'https://www.wikipedia.org/',
        'https://yandex.ru/',
        'https://yahoo.com/',
        'https://www.whatsapp.com/'
        ]
    # check the status of all websites
    for url in sites:
        # get the status for the url
        status = await get_status(url)
        # report the url and its status
        print(f'{url:30}:t{status}')
 
# run the asyncio program
asyncio.run(main())

執行範例首先建立 main() 協程並將其用作程式的入口點。main() 協程執行，定義前 10 個網站的列表。然後順序遍歷網站列表。 main()協程掛起呼叫get_status()協程查詢一個網站的狀態。

get_status() 協程執行、解析 URL 並開啟連線。它構造一個 HTTP GET 查詢並將其寫入主機。讀取、解碼並返回響應。main() 協程恢復並報告 URL 的 HTTP 狀態。

對列表中的每個 URL 重複此操作。該程式大約需要 5.6 秒才能完成，或者平均每個 URL 大約需要半秒。這突出了我們如何使用 asyncio 來查詢網頁的 HTTP 狀態。

儘管如此，它並沒有充分利用 asyncio 來並行執行任務。

https://www.google.com/ :   HTTP/1.1 200 OK
https://www.youtube.com/ :   HTTP/1.1 200 OK
https://www.facebook.com/ :   HTTP/1.1 302 Found
https://twitter.com/ :   HTTP/1.1 200 OK
https://www.instagram.com/ :   HTTP/1.1 200 OK
https://www.baidu.com/ :   HTTP/1.1 200 OK
https://www.wikipedia.org/ :   HTTP/1.1 200 OK
https://yandex.ru/ :   HTTP/1.1 302 Moved temporarily
https://yahoo.com/ :   HTTP/1.1 301 Moved Permanently
https://www.whatsapp.com/ :   HTTP/1.1 302 Found

7. 並行檢視網站狀態範例

asyncio 的一個好處是我們可以同時執行許多協程。我們可以使用 asyncio.gather() 函數在 asyncio 中並行查詢網站的狀態。

此函數採用一個或多個協程，暫停執行提供的協程，並將每個協程的結果作為可迭代物件返回。然後我們可以遍歷 URL 列表和可迭代的協程返回值並報告結果。

這可能是比上述方法更簡單的方法。首先，我們可以建立一個協程列表。

...
# create all coroutine requests
coros = [get_status(url) for url in sites]

接下來，我們可以執行協程並使用 asyncio.gather() 獲取可迭代的結果。

請注意，我們不能直接提供協程列表，而是必須將列表解壓縮為單獨的表示式，這些表示式作為位置引數提供給函數。

...
# execute all coroutines and wait
results = await asyncio.gather(*coros)

這將同時執行所有協程並檢索它們的結果。然後我們可以遍歷 URL 列表和返回狀態並依次報告每個。

...
# process all results
for url, status in zip(sites, results):
    # report status
    print(f'{url:30}:t{status}')

將它們結合在一起，下面列出了完整的範例。

# SuperFastPython.com
# check the status of many webpages
import asyncio
from urllib.parse import urlsplit
 
# get the HTTP/S status of a webpage
async def get_status(url):
    # split the url into components
    url_parsed = urlsplit(url)
    # open the connection
    if url_parsed.scheme == 'https':
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 443, ssl=True)
    else:
        reader, writer = await asyncio.open_connection(url_parsed.hostname, 80)
    # send GET request
    query = f'GET {url_parsed.path} HTTP/1.1rnHost: {url_parsed.hostname}rnrn'
    # write query to socket
    writer.write(query.encode())
    # wait for the bytes to be written to the socket
    await writer.drain()
    # read the single line response
    response = await reader.readline()
    # close the connection
    writer.close()
    # decode and strip white space
    status = response.decode().strip()
    # return the response
    return status
 
# main coroutine
async def main():
    # list of top 10 websites to check
    sites = ['https://www.google.com/',
        'https://www.youtube.com/',
        'https://www.facebook.com/',
        'https://twitter.com/',
        'https://www.instagram.com/',
        'https://www.baidu.com/',
        'https://www.wikipedia.org/',
        'https://yandex.ru/',
        'https://yahoo.com/',
        'https://www.whatsapp.com/'
        ]
    # create all coroutine requests
    coros = [get_status(url) for url in sites]
    # execute all coroutines and wait
    results = await asyncio.gather(*coros)
    # process all results
    for url, status in zip(sites, results):
        # report status
        print(f'{url:30}:t{status}')
 
# run the asyncio program
asyncio.run(main())

執行該範例會像以前一樣執行 main() 協程。在這種情況下，協程列表是在列表理解中建立的。

然後呼叫 asyncio.gather() 函數，傳遞協程並掛起 main() 協程，直到它們全部完成。協程執行，同時查詢每個網站並返回它們的狀態。

main() 協程恢復並接收可迭代的狀態值。然後使用 zip() 內建函數遍歷此可迭代物件和 URL 列表，並報告狀態。

這突出了一種更簡單的方法來同時執行協程並在所有任務完成後報告結果。它也比上面的順序版本更快，在我的系統上完成大約 1.4 秒。