搜尋

首頁 >後端開發 >Python教學 >如何使用Python Asyncio實現網站狀態檢查

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

PHPz
PHPz轉載
2023-04-21 14:10:18828瀏覽

我們可以透過開啟流並寫入和讀取 HTTP 請求和回應來使用 asyncio 查詢網站的 HTTP 狀態。

然後我們可以使用 asyncio 並發查詢多個網站的狀態，甚至是動態報告結果。

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

重要的是，每行末尾必須有一個回車和一個換行符（\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()

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.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

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.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實現網站狀態檢查的詳細內容。更多資訊請關注PHP中文網其他相關文章！

Python html 字符串 循环 并发 对象 http https ssl
陳述：
本文轉載於：yisu.com。如有侵權，請聯絡admin@php.cn刪除
上一篇：如何用Python批量刪除或移動指定影像？下一篇：如何用Python批量刪除或移動指定影像？

相關文章

看更多