关于程序员:Python-异步-检查网站状态示例21

动动发财的小手，点个赞吧！

咱们能够通过关上流并写入和读取 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

本文由 mdnice 多平台公布

关于程序员:Python-异步-检查网站状态示例21

1. 如何应用 Asyncio 查看 HTTP 状态

2. 关上 HTTP 连贯

3. 写入 HTTP 申请

4. 读取 HTTP 响应

5. 敞开 HTTP 连贯

6. 程序查看 HTTP 状态的示例

7. 并发查看网站状态示例

Just My Socks（注册教程内含优惠码）

关于程序员:Python-异步-检查网站状态示例21

1. 如何应用 Asyncio 查看 HTTP 状态

2. 关上 HTTP 连贯

3. 写入 HTTP 申请

4. 读取 HTTP 响应

5. 敞开 HTTP 连贯

6. 程序查看 HTTP 状态的示例

7. 并发查看网站状态示例

Just My Socks（注册教程 内含优惠码）

Just My Socks（注册教程内含优惠码）