溫馨提示×
您好，登錄后才能下訂單哦！
點(diǎn)擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
本篇內(nèi)容介紹了“怎么使用Python Asyncio實(shí)現(xiàn)網(wǎng)站狀態(tài)檢查”的有關(guān)知識，在實(shí)際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領(lǐng)大家學(xué)習(xí)一下如何處理這些情況吧！希望大家仔細(xì)閱讀，能夠?qū)W有所成！
asyncio 模塊提供了對打開套接字連接和通過流讀寫數(shù)據(jù)的支持。我們可以使用此功能來檢查網(wǎng)頁的狀態(tài)。
這可能涉及四個步驟，它們是：
打開一個連接
寫一個請求
讀取響應(yīng)
關(guān)閉連接
可以使用 asyncio.open_connection() 函數(shù)在 asyncio 中打開連接。在眾多參數(shù)中，該函數(shù)采用字符串主機(jī)名和整數(shù)端口號。
這是一個必須等待的協(xié)程，它返回一個 StreamReader 和一個 StreamWriter，用于使用套接字進(jìn)行讀寫。
這可用于在端口 80 上打開 HTTP 連接。
...
# open a socket connection
reader, writer = await asyncio.open_connection('www.google.com', 80)我們還可以使用 ssl=True 參數(shù)打開 SSL 連接。這可用于在端口 443 上打開 HTTPS 連接。
...
# open a socket connection
reader, writer = await asyncio.open_connection('www.google.com', 443)打開后，我們可以向 StreamWriter 寫入查詢以發(fā)出 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 之前，此字符串必須編碼為字節(jié)。這可以通過對字符串本身使用 encode() 方法來實(shí)現(xiàn)。默認(rèn)的“utf-8”編碼可能就足夠了。
... # encode string as bytes byte_data = string.encode()
然后可以通過 StreamWriter 的 write() 方法將字節(jié)寫入套接字。
... # write query to socket writer.write(byte_data)
寫入請求后，最好等待字節(jié)數(shù)據(jù)發(fā)送完畢并等待套接字準(zhǔn)備就緒。這可以通過 drain() 方法來實(shí)現(xiàn)。這是一個必須等待的協(xié)程。
... # wait for the socket to be ready. await writer.drain()
發(fā)出 HTTP 請求后，我們可以讀取響應(yīng)。這可以通過套接字的 StreamReader 來實(shí)現(xiàn)。可以使用讀取一大塊字節(jié)的 read() 方法或讀取一行字節(jié)的 readline() 方法來讀取響應(yīng)。
我們可能更喜歡 readline() 方法，因?yàn)槲覀兪褂玫氖腔谖谋镜?HTTP 協(xié)議，它一次發(fā)送一行 HTML 數(shù)據(jù)。readline() 方法是協(xié)程，必須等待。
... # read one line of response line_bytes = await reader.readline()
HTTP 1.1 響應(yīng)由兩部分組成，一個由空行分隔的標(biāo)頭，然后是一個空行終止的主體。header 包含有關(guān)請求是否成功以及將發(fā)送什么類型的文件的信息，body 包含文件的內(nèi)容，例如 HTML 網(wǎng)頁。
HTTP 標(biāo)頭的第一行包含服務(wù)器上所請求頁面的 HTTP 狀態(tài)。每行都必須從字節(jié)解碼為字符串。
這可以通過對字節(jié)數(shù)據(jù)使用 decode() 方法來實(shí)現(xiàn)。同樣，默認(rèn)編碼為“utf_8”。
... # decode bytes into a string line_data = line_bytes.decode()
我們可以通過關(guān)閉 StreamWriter 來關(guān)閉套接字連接。這可以通過調(diào)用 close() 方法來實(shí)現(xiàn)。
... # close the connection writer.close()
這不會阻塞并且可能不會立即關(guān)閉套接字?，F(xiàn)在我們知道如何使用 asyncio 發(fā)出 HTTP 請求和讀取響應(yīng)，讓我們看一些檢查網(wǎng)頁狀態(tài)的示例。
我們可以開發(fā)一個示例來使用 asyncio 檢查多個網(wǎng)站的 HTTP 狀態(tài)。
在此示例中，我們將首先開發(fā)一個協(xié)程來檢查給定 URL 的狀態(tài)。然后我們將為排名前 10 的網(wǎng)站中的每一個調(diào)用一次這個協(xié)程。
首先，我們可以定義一個協(xié)程，它將接受一個 URL 字符串并返回 HTTP 狀態(tài)。
# get the HTTP/S status of a webpage async def get_status(url): # ...
必須將 URL 解析為其組成部分。我們在發(fā)出 HTTP 請求時需要主機(jī)名和文件路徑。我們還需要知道 URL 方案（HTTP 或 HTTPS）以確定是否需要 SSL。
這可以使用 urllib.parse.urlsplit() 函數(shù)來實(shí)現(xiàn)，該函數(shù)接受一個 URL 字符串并返回所有 URL 元素的命名元組。
... # split the url into components url_parsed = urlsplit(url)
然后我們可以打開基于 URL 方案的 HTTP 連接并使用 URL 主機(jī)名。
... # 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)
接下來，我們可以使用主機(jī)名和文件路徑創(chuàng)建 HTTP GET 請求，并使用 StreamWriter 將編碼字節(jié)寫入套接字。
...
# 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 響應(yīng)。我們只需要包含 HTTP 狀態(tài)的響應(yīng)的第一行。
... # read the single line response response = await reader.readline()
然后可以關(guān)閉連接。
... # close the connection writer.close()
最后，我們可以解碼從服務(wù)器讀取的字節(jié)、遠(yuǎn)程尾隨空白，并返回 HTTP 狀態(tài)。
... # decode and strip white space status = response.decode().strip() # return the response return status
將它們結(jié)合在一起，下面列出了完整的 get_status() 協(xié)程。它沒有任何錯誤處理，例如無法訪問主機(jī)或響應(yīng)緩慢的情況。這些添加將為讀者提供一個很好的擴(kuò)展。
# 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接下來，我們可以為我們要檢查的多個網(wǎng)頁或網(wǎng)站調(diào)用 get_status() 協(xié)程。在這種情況下，我們將定義一個世界排名前 10 的網(wǎng)頁列表。
... # 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() 協(xié)程依次查詢每個。在這種情況下，我們將在一個循環(huán)中按順序這樣做，并依次報告每個狀態(tài)。
...
# 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 時，我們可以做得比順序更好，但這提供了一個很好的起點(diǎn)，我們可以在以后進(jìn)行改進(jìn)。將它們結(jié)合在一起，main() 協(xié)程查詢前 10 個網(wǎng)站的狀態(tài)。
# 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}')最后，我們可以創(chuàng)建 main() 協(xié)程并將其用作 asyncio 程序的入口點(diǎn)。
... # run the asyncio program asyncio.run(main())
將它們結(jié)合在一起，下面列出了完整的示例。
# 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())運(yùn)行示例首先創(chuàng)建 main() 協(xié)程并將其用作程序的入口點(diǎn)。main() 協(xié)程運(yùn)行，定義前 10 個網(wǎng)站的列表。然后順序遍歷網(wǎng)站列表。 main()協(xié)程掛起調(diào)用get_status()協(xié)程查詢一個網(wǎng)站的狀態(tài)。
get_status() 協(xié)程運(yùn)行、解析 URL 并打開連接。它構(gòu)造一個 HTTP GET 查詢并將其寫入主機(jī)。讀取、解碼并返回響應(yīng)。main() 協(xié)程恢復(fù)并報告 URL 的 HTTP 狀態(tài)。
對列表中的每個 URL 重復(fù)此操作。該程序大約需要 5.6 秒才能完成，或者平均每個 URL 大約需要半秒。這突出了我們?nèi)绾问褂?asyncio 來查詢網(wǎng)頁的 HTTP 狀態(tài)。
盡管如此，它并沒有充分利用 asyncio 來并發(fā)執(zhí)行任務(wù)。
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 的一個好處是我們可以同時執(zhí)行許多協(xié)程。我們可以使用 asyncio.gather() 函數(shù)在 asyncio 中并發(fā)查詢網(wǎng)站的狀態(tài)。
此函數(shù)采用一個或多個協(xié)程，暫停執(zhí)行提供的協(xié)程，并將每個協(xié)程的結(jié)果作為可迭代對象返回。然后我們可以遍歷 URL 列表和可迭代的協(xié)程返回值并報告結(jié)果。
這可能是比上述方法更簡單的方法。首先，我們可以創(chuàng)建一個協(xié)程列表。
... # create all coroutine requests coros = [get_status(url) for url in sites]
接下來，我們可以執(zhí)行協(xié)程并使用 asyncio.gather() 獲取可迭代的結(jié)果。
請注意，我們不能直接提供協(xié)程列表，而是必須將列表解壓縮為單獨(dú)的表達(dá)式，這些表達(dá)式作為位置參數(shù)提供給函數(shù)。
... # execute all coroutines and wait results = await asyncio.gather(*coros)
這將同時執(zhí)行所有協(xié)程并檢索它們的結(jié)果。然后我們可以遍歷 URL 列表和返回狀態(tài)并依次報告每個。
...
# process all results
for url, status in zip(sites, results):
# report status
print(f'{url:30}:\t{status}')將它們結(jié)合在一起，下面列出了完整的示例。
# 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())運(yùn)行該示例會像以前一樣執(zhí)行 main() 協(xié)程。在這種情況下，協(xié)程列表是在列表理解中創(chuàng)建的。
然后調(diào)用 asyncio.gather() 函數(shù)，傳遞協(xié)程并掛起 main() 協(xié)程，直到它們?nèi)客瓿?。協(xié)程執(zhí)行，同時查詢每個網(wǎng)站并返回它們的狀態(tài)。
main() 協(xié)程恢復(fù)并接收可迭代的狀態(tài)值。然后使用 zip() 內(nèi)置函數(shù)遍歷此可迭代對象和 URL 列表，并報告狀態(tài)。
這突出了一種更簡單的方法來同時執(zhí)行協(xié)程并在所有任務(wù)完成后報告結(jié)果。它也比上面的順序版本更快，在我的系統(tǒng)上完成大約 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實(shí)現(xiàn)網(wǎng)站狀態(tài)檢查”的內(nèi)容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識可以關(guān)注億速云網(wǎng)站，小編將為大家輸出更多高質(zhì)量的實(shí)用文章！
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號
手機(jī)網(wǎng)站二維碼