溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章主要介紹了Python怎么使用Asyncio實現檢查網站狀態的相關知識，內容詳細易懂，操作簡單快捷，具有一定借鑒價值，相信大家閱讀完這篇Python怎么使用Asyncio實現檢查網站狀態文章都會有所收獲，下面我們一起來看看吧。
我們可以通過打開流并寫入和讀取 HTTP 請求和響應來使用 asyncio 查詢網站的 HTTP 狀態。
然后我們可以使用 asyncio 并發查詢多個網站的狀態，甚至動態報告結果。
asyncio 模塊提供了對打開套接字連接和通過流讀寫數據的支持。我們可以使用此功能來檢查網頁的狀態。
這可能涉及四個步驟，它們是：
打開一個連接
寫一個請求
讀取響應
關閉連接
可以使用 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)打開后，我們可以向 StreamWriter 寫入查詢以發出 HTTP 請求。例如，HTTP 版本 1.1 請求是純文本格式的。我們可以請求文件路徑“/”，它可能如下所示：
GET / HTTP/1.1 Host: www.google.com
重要的是，每行末尾必須有一個回車和一個換行符（\r\n），末尾有一個空行。
作為 Python 字符串，這可能如下所示：
'GET / HTTP/1.1\r\n' 'Host: www.google.com\r\n' '\r\n'
在寫入 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()
發出 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()
我們可以通過關閉 StreamWriter 來關閉套接字連接。這可以通過調用 close() 方法來實現。
... # close the connection writer.close()
這不會阻塞并且可能不會立即關閉套接字。現在我們知道如何使用 asyncio 發出 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.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
# 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.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
# 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.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
# 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
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.1\r\nHost: {url_parsed.hostname}\r\n\r\n'
# 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 秒。
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
關于“Python怎么使用Asyncio實現檢查網站狀態”這篇文章的內容就介紹到這里，感謝各位的閱讀！相信大家對“Python怎么使用Asyncio實現檢查網站狀態”知識都有一定的了解，大家如果還想學習更多知識，歡迎關注億速云行業資訊頻道。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
