AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

Python 异步: 检查网站状态示例（21）

2023-03-29 119
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： Python 异步: 检查网站状态示例（21）

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

我们可以通过打开流并写入和读取 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
网络安全
移动开发
关键词：
Python示例
Python异步
Python检查
Python网站
Python异步示例
冷冻工厂
目录
相关文章
龙大吉
|
2月前
|
数据挖掘 Python
Python示例，展示如何找到最近一次死叉之后尚未形成金叉的位置
【10月更文挑战第7天】金融分析中，“死叉”指短期移动平均线（如MA5）跌破长期移动平均线（如MA10），而“金叉”则相反。本文提供Python代码示例，用于找出最近一次死叉后未形成金叉的位置，涵盖移动平均线计算、交叉点判断及结果输出等步骤，适合金融数据分析。
龙大吉
45 4
技能实验室
|
1月前
|
关系型数据库 MySQL 数据库连接
python脚本：连接数据库，检查直播流是否可用
【10月更文挑战第13天】本脚本使用 `mysql-connector-python` 连接MySQL数据库，检查 `live_streams` 表中每个直播流URL的可用性。通过 `requests` 库发送HTTP请求，输出每个URL的检查结果。需安装 `mysql-connector-python` 和 `requests` 库，并配置数据库连接参数。
技能实验室
128 68
蓝易云
|
7天前
|
数据可视化 Python
以下是一些常用的图表类型及其Python代码示例，使用Matplotlib和Seaborn库。
通过这些思维导图和分析说明表，您可以更直观地理解和选择适合的数据可视化图表类型，帮助更有效地展示和分析数据。
蓝易云
45 8
路边两盏灯
|
14天前
|
API Python
【Azure Developer】分享一段Python代码调用Graph API创建用户的示例
分享一段Python代码调用Graph API创建用户的示例
路边两盏灯
38 11
左诗右码
|
15天前
|
数据采集 JSON 测试技术
Grequests，非常 Nice 的 Python 异步 HTTP 请求神器
在Python开发中，处理HTTP请求至关重要。`grequests`库基于`requests`，支持异步请求，通过`gevent`实现并发，提高性能。本文介绍了`grequests`的安装、基本与高级功能，如GET/POST请求、并发控制等，并探讨其在实际项目中的应用。
左诗右码
24 3
bruce_xiaowei
|
17天前
|
网络安全 Python
Python网络编程小示例：生成CIDR表示的IP地址范围
本文介绍了如何使用Python生成CIDR表示的IP地址范围，通过解析CIDR字符串，将其转换为二进制形式，应用子网掩码，最终生成该CIDR块内所有可用的IP地址列表。示例代码利用了Python的`ipaddress`模块，展示了从指定CIDR表达式中提取所有IP地址的过程。
bruce_xiaowei
35 6
龙大吉
|
1月前
|
数据挖掘 Python
Python示例，展示如何找到最近一次死叉之后尚未形成金叉的位置
金融分析中，“死叉”指短期移动平均线（如MA5）跌破长期移动平均线（如MA10），而“金叉”则相反。本文提供Python代码示例，用于找出最近一次死叉后未形成金叉的位置，涵盖移动平均线计算、交叉点判断及结果输出等步骤，适合金融数据分析。
龙大吉
23 1
aliyun0641063962-48771
|
2月前
|
关系型数据库 MySQL 数据处理
探索Python中的异步编程：从asyncio到异步数据库操作
在这个快节奏的技术世界里，效率和性能是关键。本文将带你深入Python的异步编程世界，从基础的asyncio库开始，逐步探索到异步数据库操作的高级应用。我们将一起揭开异步编程的神秘面纱，探索它如何帮助我们提升应用程序的性能和响应速度。
aliyun0641063962-48771
64 3
AIweker
|
2月前
|
调度 Python
深入理解 Python 中的异步操作 | python小知识
在现代编程中，异步操作是一个非常重要的概念，尤其是在处理 I/O 密集型任务时。使用异步操作可以显著提高程序的性能和响应速度。Python 提供了 `async` 和 `await` 关键字，使得编写异步代码变得更加直观和简洁【10月更文挑战第8天】
AIweker
31 2
墨城coding
|
1月前
|
NoSQL 关系型数据库 MySQL
python协程+异步总结！
本文介绍了Python中的协程、asyncio模块以及异步编程的相关知识。首先解释了协程的概念和实现方法，包括greenlet、yield关键字、asyncio装饰器和async/await关键字。接着详细讲解了协程的意义和应用场景，如提高IO密集型任务的性能。文章还介绍了事件循环、Task对象、Future对象等核心概念，并提供了多个实战案例，包括异步Redis、MySQL操作、FastAPI框架和异步爬虫。最后提到了uvloop作为asyncio的高性能替代方案。通过这些内容，读者可以全面了解和掌握Python中的异步编程技术。
墨城coding
47 0

热门文章

最新文章

  • 1
    3、Python与设计模式--建造者模式
  • 2
    使用cProfile等工具来提高python的执行速度
  • 3
    2022年全国大学生数学建模竞赛E题目-小批量物料生产安排详解+思路+Python代码时序预测模型(三)
  • 4
    python日志模块logging
  • 5
    python: tk demo
  • 6
    Python 包管理（pip、conda）基本使用指南
  • 7
    python中math模块常用的方法整理
  • 8
    练习PYTHON协程之GREENLET
  • 9
    构建Python软件大厦系列
  • 10
    初学者必知的Python中优雅的用法
  • 1
    Python 密码学实用指南（全）（2）
    36
  • 2
    【python】习题 6-10周（下）
    57
  • 3
    【python】习题 6-10周（中）
    65
  • 4
    Python 入门指南（七）（4）
    48
  • 5
    【python】习题 1-5周（中）
    83
  • 6
    【python】习题 1-5周（上）
    50
  • 7
    Python 入门指南（七）（2）
    55
  • 8
    Python 入门指南（七）（1）
    41
  • 9
    Python 入门指南（六）（4）
    28
  • 10
    Python 入门指南（六）（3）
    41

    • 相关课程

    更多
  • Python Web开发基础
  • Python开发基础入门
  • Python常用数据科学库
  • Python网络爬虫实战
  • Python完全自学手册图文教程
  • Python基础快速入门实战教程

    • 相关电子书

    更多
  • From Python Scikit-Learn to Sc
  • Data Pre-Processing in Python:
  • 双剑合璧-Python和大数据计算平台的结合

    • 相关实验场景

    更多
  • 用Python画圣诞树
  • Python选择及循环结构
  • 云端Python及基本操作
  • Python网络通信程序典型应用
  • Python新手入门（Anolis OS）
  • Python新手入门

    • 推荐镜像

    更多
  • python-release
  • nodejs-release
  • debian-cd
    • 下一篇
    fs.oss.accessKeyId和fs.oss.accessKeySecret。