讲师中心 微信公众号
首页
文章
后端开发 web前端 数据库 开发工具 php框架 常见问题 人工智能 Java 系统教程 电脑教程 硬件教程 手机教程 软件教程 游戏教程 自媒体 新闻
专题
后端开发 web前端 数据库 开发工具 php框架 人工智能 Java 系统教程 电脑教程 硬件教程 手机教程 软件教程 游戏教程 新闻
AI工具
AI聊天问答 Agent智能体 AI文本写作 AI绘画作图 AI设计工具 AI视频创作 AI音频制作 AI办公学习 AI编程开发 AI提示词
学习
大前端 后端开发 数据库 移动端 运维开发 计算机基础
编程手册
大前端 后端开发 数据库 移动端 运维开发 计算机基础
下载
js特效 网站源码 工具下载 类库下载 网站素材 学习资源 插件扩展 手机游戏
最近更新
当前位置：首页 > 后端开发 > Python教程 >

正文

0

0

如何使用Python Asyncio实现网站状态检查

PHPz

PHPz

发布时间：2023-04-21 14:10:18

|

1090人浏览过

|

来源于亿速云

转载

我们可以通过打开流并写入和读取 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 状态。

白月生产企业订单管理系统GBK2.0 Build 080807
白月生产企业订单管理系统GBK2.0 Build 080807

请注意以下说明:1、本程序允许任何人免费使用。2、本程序采用PHP+MYSQL架构编写。并且经过ZEND加密，所以运行环境需要有ZEND引擎支持。3、需要售后服务的，请与本作者联系，联系方式见下方。4、本程序还可以与您的网站想整合，可以实现用户在线服务功能，可以让客户管理自己的信息，可以查询自己的订单状况。以及返点信息等相关客户利益的信息。这个功能可提高客户的向心度。安装方法：1、解压本系统，放在

下载 
# 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 OKhttps://www.youtube.com/      :    HTTP/1.1 200 OKhttps://www.facebook.com/     :    HTTP/1.1 302 Foundhttps://twitter.com/          :    HTTP/1.1 200 OKhttps://www.instagram.com/    :    HTTP/1.1 200 OKhttps://www.baidu.com/        :    HTTP/1.1 200 OKhttps://www.wikipedia.org/    :    HTTP/1.1 200 OKhttps://yandex.ru/            :    HTTP/1.1 302 Moved temporarilyhttps://yahoo.com/            :    HTTP/1.1 301 Moved Permanentlyhttps://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 OKhttps://www.youtube.com/      :    HTTP/1.1 200 OKhttps://www.facebook.com/     :    HTTP/1.1 302 Foundhttps://twitter.com/          :    HTTP/1.1 200 OKhttps://www.instagram.com/    :    HTTP/1.1 200 OKhttps://www.baidu.com/        :    HTTP/1.1 200 OKhttps://www.wikipedia.org/    :    HTTP/1.1 200 OKhttps://yandex.ru/            :    HTTP/1.1 302 Moved temporarilyhttps://yahoo.com/            :    HTTP/1.1 301 Moved Permanentlyhttps://www.whatsapp.com/     :    HTTP/1.1 302 Found

相关文章

Python进程资源回收_进程退出清理机制

Python系统如何做容灾设计_高可用架构

Python内存分析工具_tracemalloc讲解

Python 中 else 语法错误的典型原因与修复方法

将字符串安全转换为浮点数并参与数值计算的正确方法
python速学教程(入门到精通)
python速学教程(入门到精通)

python怎么学习？python怎么入门？python在哪学？python怎么学才快？不用担心，这里为大家提供了python速学教程(入门到精通)，有需要的小伙伴保存下载就能学习啦！

下载

相关标签:

python 字符串 循环 并发 对象 http https ssl

本站声明：本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系admin@php.cn

上一篇：如何用Python批量删除或移动指定图像？ 下一篇：Python字符串：常见用法与f-string源码分析

作者最新文章

如何高效生成订单/发票号？SprykerSequenceNumber模块助你轻松搞定

2025-09-16 10:01

如何解决电商平台商品属性管理混乱的问题，使用SprykerProductAttribute模块助你实现灵活高效的数据管理

2025-09-16 12:23

解锁夸克浏览器AI搜索新功能_掌握夸克AI搜索的进阶玩法

2025-10-13 17:08

升级夸克浏览器体验AI搜索_夸克AI搜索核心功能深度解析

2025-10-28 20:58

微信朋友圈能不能定时发 微信朋友圈定时发送辅助工具使用

2026-01-11 08:41

企业微信朋友圈怎么定时发送 企业微信定时发布朋友圈教程

2026-01-19 01:11

微信朋友圈怎么预约发送 微信朋友圈定时发布最新解决方案

2026-02-09 10:52

mysql如何进行逻辑运算_mysql and/or/not组合

2026-03-06 10:26

mysql如何使用位运算符_mysql二进制数据处理

2026-03-07 02:17

微信怎么隐藏好友如何单独隐藏一个人不删除不拉黑

2026-03-10 10:02

热门AI工具

更多
DeepSeek
DeepSeek

幻方量化公司旗下的开源大模型平台

AI编程开发AI聊天问答
豆包大模型
豆包大模型

字节跳动自主研发的一系列大型语言模型

AI编程开发AI大模型
通义千问
通义千问

阿里巴巴推出的全能AI助手

AI编程开发Agent智能体
腾讯元宝
腾讯元宝

腾讯混元平台推出的AI助手

文档处理Excel 表格
文心一言
文心一言

文心一言是百度开发的AI聊天机器人，通过对话可以生成各种形式的内容。

AI编程开发AI文本写作
讯飞写作
讯飞写作

基于讯飞星火大模型的AI写作工具，可以快速生成新闻稿件、品宣文案、工作总结、心得体会等各种文文稿

AI文本写作中文写作
即梦AI
即梦AI

一站式AI创作平台，免费AI图片和视频生成。

图片拼接图画生成
ChatGPT
ChatGPT

最最强大的AI聊天机器人程序，ChatGPT不单是聊天机器人，还能进行撰写邮件、视频脚本、文案、翻译、代码等任务。

AI编程开发AI文本写作
智谱清言 - 免费全能的AI助手
智谱清言 - 免费全能的AI助手

智谱清言 - 免费全能的AI助手

AI编程开发Agent智能体

相关专题

更多
js 字符串转数组
js 字符串转数组

js字符串转数组的方法：1、使用“split()”方法；2、使用“Array.from()”方法；3、使用for循环遍历；4、使用“Array.split()”方法。本专题为大家提供js字符串转数组的相关的文章、下载、课程内容，供大家免费下载体验。

739

2023.08.03

js截取字符串的方法
js截取字符串的方法

js截取字符串的方法有substring()方法、substr()方法、slice()方法、split()方法和slice()方法。本专题为大家提供字符串相关的文章、下载、课程内容，供大家免费下载体验。

220

2023.09.04

java基础知识汇总
java基础知识汇总

java基础知识有Java的历史和特点、Java的开发环境、Java的基本数据类型、变量和常量、运算符和表达式、控制语句、数组和字符串等等知识点。想要知道更多关于java基础知识的朋友，请阅读本专题下面的的有关文章，欢迎大家来php中文网学习。

1564

2023.10.24

字符串介绍
字符串介绍

字符串是一种数据类型，它可以是任何文本，包括字母、数字、符号等。字符串可以由不同的字符组成，例如空格、标点符号、数字等。在编程中，字符串通常用引号括起来，如单引号、双引号或反引号。想了解更多字符串的相关内容，可以阅读本专题下面的文章。

649

2023.11.24

java读取文件转成字符串的方法
java读取文件转成字符串的方法

Java8引入了新的文件I/O API，使用java.nio.file.Files类读取文件内容更加方便。对于较旧版本的Java，可以使用java.io.FileReader和java.io.BufferedReader来读取文件。在这些方法中，你需要将文件路径替换为你的实际文件路径，并且可能需要处理可能的IOException异常。想了解更多java的相关内容，可以阅读本专题下面的文章。

1208

2024.03.22

php中定义字符串的方式
php中定义字符串的方式

php中定义字符串的方式：单引号；双引号；heredoc语法等等。想了解更多字符串的相关内容，可以阅读本专题下面的文章。

1184

2024.04.29

go语言字符串相关教程
go语言字符串相关教程

本专题整合了go语言字符串相关教程，阅读专题下面的文章了解更多详细内容。

191

2025.07.29

c++字符串相关教程
c++字符串相关教程

本专题整合了c++字符串相关教程，阅读专题下面的文章了解更多详细内容。

111

2025.08.07

Go高并发任务调度与Goroutine池化实践
Go高并发任务调度与Goroutine池化实践

本专题围绕 Go 语言在高并发任务处理场景中的实践展开，系统讲解 Goroutine 调度模型、Channel 通信机制以及并发控制策略。内容包括任务队列设计、Goroutine 池化管理、资源限制控制以及并发任务的性能优化方法。通过实际案例演示，帮助开发者构建稳定高效的 Go 并发任务处理系统，提高系统在高负载环境下的处理能力与稳定性。

4

2026.03.10

热门下载

更多
网站特效
/
网站源码
/
网站素材
/
前端模板

相关下载

更多
白月生产企业订单管理系统GBK2.0 Build 080807

精品课程

更多
相关推荐
/
热门推荐
/
最新课程
最新Python教程 从入门到精通
最新Python教程 从入门到精通

共4课时 | 22.5万人学习

Django 教程
Django 教程

共28课时 | 4.9万人学习

SciPy 教程
SciPy 教程

共10课时 | 1.9万人学习

最新文章

更多
Python属性访问控制_property用法
如何在 Pandas 中按分组保留每组日期最大值对应的整行数据
使用 Pandas 实现条件重置的累积求和（类似 Excel 的递推公式）
如何在 MLRun 中安全高效地读取已存储的 CSV 类型 Artifact
Python事件循环解析_event_loop说明
如何在 Pandas 中基于当前行、前一行及新列自身历史值动态生成列
Python如何排查线上故障_日志定位方法
如何在 Pandas 中按分组保留日期最新的行
如何正确将混合类型数据（含数字字符串）转换为浮点数并计算平均值
Python进程资源回收_进程退出清理机制
关于我们 免责申明 举报中心 意见反馈 讲师合作 广告合作 最新更新
php中文网：公益在线php培训，帮助PHP学习者快速成长！
关注服务号 技术交流群
PHP中文网订阅号
每天精选资源文章推送

Copyright 2014-2026 https://www.php.cn/ All Rights Reserved | php.cn | 湘ICP备2023035733号

  • PHP学习

  • 技术支持

  • 返回顶部