Python 异步协程：从 async/await 到 asyncio 再到 async with

news/2024/12/24 6:12:40/

在 Python 3.8 以后的版本中，异步编程变得越来越重要。本文将系统介绍 Python 标准库中的异步编程工具，带领大家掌握 async/await 语法和 asyncio 的使用。

从一个简单的场景开始

假设我们在处理一些耗时的 I/O 操作，比如读取多个文件或处理多个数据。为了模拟这种场景，我们先用 time.sleep() 来代表耗时操作：

python">import time
import randomdef process_item(item):# 模拟耗时操作print(f"处理中：{item}")process_time = random.uniform(0.5, 2.0)time.sleep(process_time)return f"处理完成：{item}，耗时 {process_time:.2f} 秒"def process_all_items():items = ["任务A", "任务B", "任务C", "任务D"]results = []for item in items:result = process_item(item)results.append(result)return resultsif __name__ == "__main__":start = time.time()results = process_all_items()end = time.time()print("\n".join(results))print(f"总耗时：{end - start:.2f} 秒")

处理中：任务A
处理中：任务B
处理中：任务C
处理中：任务D
处理完成：任务A，耗时 1.97 秒
处理完成：任务B，耗时 1.28 秒
处理完成：任务C，耗时 0.66 秒
处理完成：任务D，耗时 1.80 秒
总耗时：5.72 秒

这段代码的问题很明显：每个任务都必须等待前一个任务完成才能开始。如果有4个任务，每个任务平均耗时1秒，那么总耗时就接近4秒。

认识 async/await

Python 引入了 async/await 语法来支持异步编程。当我们在函数定义前加上 async 关键字时，这个函数就变成了一个"协程"（coroutine）。而 await 关键字则用于等待一个协程完成。让我们改写上面的代码：

python">import asyncio
import random
import timeasync def process_item(item):print(f"处理中：{item}")# async 定义的函数变成了协程process_time = random.uniform(0.5, 2.0)# time.sleep() 换成 asyncio.sleep()await asyncio.sleep(process_time)  # await 等待异步操作完成return f"处理完成：{item}，耗时 {process_time:.2f} 秒"async def process_all_items():items = ["任务A", "任务B", "任务C", "任务D"]# 创建任务列表tasks = [asyncio.create_task(process_item(item))for item in items]print("开始处理")results = await asyncio.gather(*tasks)return resultsasync def main():start = time.time()results = await process_all_items()end = time.time()print("\n".join(results))print(f"总耗时：{end - start:.2f} 秒")if __name__ == "__main__":asyncio.run(main())

开始处理
处理中：任务A
处理中：任务B
处理中：任务C
处理中：任务D
处理完成：任务A，耗时 1.97 秒
处理完成：任务B，耗时 0.80 秒
处理完成：任务C，耗时 0.83 秒
处理完成：任务D，耗时 1.46 秒
总耗时：1.97 秒

让我们详细解释这段代码的执行过程：

  1. 当函数被 async 关键字修饰后，调用该函数不会直接执行函数体，而是返回一个协程对象
  2. await 关键字只能在 async 函数内使用，它表示"等待这个操作完成后再继续"
  3. asyncio.create_task() 将协程包装成一个任务，该任务会被事件循环调度执行
  4. asyncio.gather() 并发运行多个任务，并等待它们全部完成
  5. asyncio.run() 创建事件循环，运行 main() 协程，直到它完成

使用 asyncio.wait_for 添加超时控制

在实际应用中，我们往往需要为异步操作设置超时时间：

python">import asyncio
import random
import timeasync def process_item(item):process_time = random.uniform(0.5, 2.0)try:# 设置1秒超时await asyncio.wait_for(asyncio.sleep(process_time),timeout=1.0)return f"处理完成：{item}，耗时 {process_time:.2f} 秒"except asyncio.TimeoutError:return f"处理超时：{item}"async def main():items = ["任务A", "任务B", "任务C", "任务D"]tasks = [asyncio.create_task(process_item(item))for item in items]start = time.time()results = await asyncio.gather(*tasks, return_exceptions=True)end = time.time()print("\n".join(results))print(f"总耗时：{end - start:.2f} 秒")if __name__ == "__main__":asyncio.run(main())

处理超时：任务A
处理完成：任务B，耗时 0.94 秒
处理超时：任务C
处理完成：任务D，耗时 0.78 秒
总耗时：1.00 秒

使用异步上下文管理器

Python 中的 with 语句可以用于资源管理，类似地，异步编程中我们可以使用 async with 。一个类要支持异步上下文管理，需要实现 __aenter__ 和 __aexit__ 方法：

python">import asyncio
import randomclass AsyncResource:async def __aenter__(self):# 异步初始化资源print("正在初始化资源...")await asyncio.sleep(0.1)return selfasync def __aexit__(self, exc_type, exc_val, exc_tb):# 异步清理资源print("正在清理资源...")await asyncio.sleep(0.1)async def process(self, item):# 异步处理任务print(f"正在处理任务：{item}")process_time = random.uniform(0.5, 2.0)await asyncio.sleep(process_time)return f"处理完成：{item}，耗时 {process_time:.2f} 秒"async def main():items = ["任务A", "任务B", "任务C"]async with AsyncResource() as resource:tasks = [asyncio.create_task(resource.process(item))for item in items]results = await asyncio.gather(*tasks)print("\n".join(results))if __name__ == "__main__":asyncio.run(main())

正在初始化资源...
正在处理任务：任务A
正在处理任务：任务B
正在处理任务：任务C
正在清理资源...
处理完成：任务A，耗时 1.31 秒
处理完成：任务B，耗时 0.77 秒
处理完成：任务C，耗时 0.84 秒

使用事件循环执行阻塞操作 run_in_executor

在异步编程中，我们可能会遇到一些无法避免的阻塞操作（比如调用传统的同步API）。这时，asyncio.get_running_loop() 和 run_in_executor 就显得特别重要：

python">import asyncio
import time
import requests  # 一个同步的HTTP客户端库async def blocking_operation():# 获取当前事件循环loop = asyncio.get_running_loop()# 在线程池中执行阻塞操作result = await loop.run_in_executor(None,  # 使用默认的线程池执行器requests.get,  # 要执行的阻塞函数'http://httpbin.org/delay/1'  # 函数参数)return result.status_codeasync def non_blocking_operation():await asyncio.sleep(1)return "非阻塞操作完成"async def main():# 同时执行阻塞和非阻塞操作tasks = [asyncio.create_task(blocking_operation()),asyncio.create_task(non_blocking_operation())]start = time.time()results = await asyncio.gather(*tasks)end = time.time()print(f"操作结果：{results}")print(f"总耗时：{end - start:.2f} 秒")if __name__ == "__main__":asyncio.run(main())

输出：

操作结果：[200, '非阻塞操作完成']
总耗时：1.99 秒

这个例子展示了如何在异步程序中优雅地处理同步操作。如果不使用 run_in_executor，阻塞操作会阻塞整个事件循环，导致其他任务无法执行：

  • requests.get() 是同步操作,会阻塞当前线程
  • 事件循环运行在主线程上
  • 如果直接在协程中调用 requests.get() ，整个事件循环都会被阻塞
  • 其他任务无法在这期间执行
  • run_in_executor 会将阻塞操作放到另一个线程中执行
  • 主线程的事件循环可以继续处理其他任务
  • 当线程池中的操作完成时，结果会被返回给事件循环

最佳实践是：

  • 尽量使用原生支持异步的库(如 aiohttp)
  • 如果必须使用同步库，就用 run_in_executor
  • 对于 CPU 密集型任务也可以用 run_in_executor 放到进程池中执行

任务取消：优雅地终止异步操作

有时我们需要取消正在执行的异步任务，比如用户中断操作或超时处理：

python">import asyncio
import randomasync def long_operation(name):try:print(f"{name} 开始执行")while True:  # 模拟一个持续运行的操作await asyncio.sleep(0.5)print(f"{name} 正在执行...")except asyncio.CancelledError:print(f"{name} 被取消了")raise  # 重要：继续传播取消信号async def main():# 创建三个任务task1 = asyncio.create_task(long_operation("任务1"))task2 = asyncio.create_task(long_operation("任务2"))task3 = asyncio.create_task(long_operation("任务3"))# 等待1秒后取消task1await asyncio.sleep(1)task1.cancel()# 等待2秒后取消其余任务await asyncio.sleep(1)task2.cancel()task3.cancel()try:# 等待所有任务完成或被取消await asyncio.gather(task1, task2, task3, return_exceptions=True)except asyncio.CancelledError:print("某个任务被取消了")if __name__ == "__main__":asyncio.run(main())

输出：

任务1 开始执行
任务2 开始执行
任务3 开始执行
任务1 正在执行...
任务2 正在执行...
任务3 正在执行...
任务1 被取消了
任务2 正在执行...
任务3 正在执行...
任务2 正在执行...
任务3 正在执行...
任务2 被取消了
任务3 被取消了

这个例子展示了如何正确处理任务取消：

  1. 任务可以在执行过程中被取消
  2. 被取消的任务会抛出 CancelledError
  3. 我们应该适当处理取消信号，确保资源被正确清理

深入理解协程：为什么需要 async/await？

协程（Coroutine）是一种特殊的函数，它可以在执行过程中暂停，并在之后从暂停的地方继续执行。当我们使用 async 定义一个函数时，我们实际上是在定义一个协程：

python">import asyncio# 这是一个普通函数
def normal_function():return "Hello"# 这是一个协程
async def coroutine_function():await asyncio.sleep(1)return "Hello"# 让我们看看它们的区别
print(normal_function)      # <function normal_function at 0x1052cc040>
print(coroutine_function)   # <function coroutine_function at 0x1054b9790># 调用它们的结果不同
print(normal_function())    # 直接返回: "Hello"
print(coroutine_function()) # RuntimeWarning: coroutine 'coroutine_function' was never awaited
# <coroutine object coroutine_function at 0x105962e40>

await 如何与事件循环协作

协程（Coroutine）的核心在于它可以在执行过程中主动交出控制权，让其他代码有机会执行。让我们通过一个详细的例子来理解这个过程：

python">import asyncioasync def task1():print("任务1：开始")print("任务1：准备休眠")await asyncio.sleep(2)  # 关键点1：交出控制权print("任务1：休眠结束")async def task2():print("任务2：开始")print("任务2：准备休眠")await asyncio.sleep(1)  # 关键点2：交出控制权print("任务2：休眠结束")async def main():# 同时执行两个任务await asyncio.gather(task1(), task2())asyncio.run(main())

这段代码的输出会是：

任务1：开始
任务1：准备休眠
任务2：开始
任务2：准备休眠
任务2：休眠结束    # 1秒后
任务1：休眠结束    # 2秒后

让我们详细解释执行过程：

  1. 当程序遇到 await asyncio.sleep(2) 时：

    • 这个 sleep 操作被注册到事件循环中
    • Python 记录当前的执行位置
    • task1 主动交出控制权
    • 重要：task1 并没有停止运行，而是被暂停了，等待之后恢复

  2. 事件循环接管控制权后：

    • 寻找其他可以执行的协程（这里是 task2）
    • 开始执行 task2，直到遇到 await asyncio.sleep(1)
    • task2 也交出控制权，被暂停

  3. 事件循环继续工作：

    • 管理一个计时器，追踪这两个 sleep 操作
    • 1秒后，发现 task2 的 sleep 时间到了
    • 恢复 task2 的执行，打印"任务2：休眠结束"
    • 2秒到时，恢复 task1 的执行，打印"任务1：休眠结束"

这就像是一个指挥家（事件循环）在指挥一个管弦乐队（多个协程）：

  • 当某个乐器（协程）需要休息时，它举手示意（await）
  • 指挥家看到后，立即指挥其他乐器演奏
  • 当休息时间到了，指挥家会示意这个乐器继续演奏

代码验证：

python">import asyncio
import timeasync def report_time(name, sleep_time):print(f"{time.strftime('%H:%M:%S')} - {name}开始")await asyncio.sleep(sleep_time)print(f"{time.strftime('%H:%M:%S')} - {name}结束")async def main():# 同时执行多个任务await asyncio.gather(report_time("任务A", 2),report_time("任务B", 1),report_time("任务C", 3))asyncio.run(main())

输出：

00:19:26 - 任务A开始
00:19:26 - 任务B开始
00:19:26 - 任务C开始
00:19:27 - 任务B结束
00:19:28 - 任务A结束
00:19:29 - 任务C结束

这种机制的优势在于：

  1. 单线程执行，没有线程切换开销
  2. 协程主动交出控制权，而不是被操作系统强制切换
  3. 比起回调地狱，代码更清晰易读
  4. 错误处理更直观，可以使用普通的 try/except

理解了这个机制，我们就能更好地使用异步编程：

  • 在 await 的时候，其他协程有机会执行
  • 耗时操作应该是真正的异步操作（比如 asyncio.sleep ）
  • 不要在协程中使用阻塞操作，那样会卡住整个事件循环

小结

Python 的异步编程主要依赖以下概念：

  1. async/await 语法：定义和等待协程
  2. asyncio 模块：提供事件循环和任务调度
  3. Task 对象：表示待执行的工作单元
  4. 异步上下文管理器：管理异步资源

使用异步编程的关键点：

  1. I/O 密集型任务最适合使用异步编程
  2. 所有耗时操作都应该是真正的异步操作
  3. 注意处理超时和异常情况
  4. 合理使用 asyncio.gather() 和 asyncio.wait_for()

异步编程不是万能的，但在处理 I/O 密集型任务时确实能带来显著的性能提升。合理使用这些工具，能让我们的程序更高效、更优雅。


http://www.ppmy.cn/news/1557660.html

相关文章

基于cobra开发的k8s命令行管理工具k8s-manager

基于cobra开发的k8s命令行管理工具k8s-manager

基于cobra开发的k8s命令行管理工具k8s-manager 如果觉得好用&#xff0c;麻烦给个Star!通用配置1 node 分析所有node的资源情况2 analysis 分析Node节点上的资源使用构成3 image 获取指定namespace的所有镜像地址4 resource 获取指定namespace的所有limit 与 Requests大小5 top…
阅读更多...
BERT模型入门（1）BERT的基本概念

BERT模型入门（1）BERT的基本概念

文章目录 BERT是Bidirectional Encoder Representations from Transformers的首字母简写&#xff0c;中文意思是&#xff1a;Transformer的双向编码器表示。它是谷歌发布的最先进的嵌入模型。BERT在许多NLP任务中提供了更好的结果&#xff0c;如问答、文本生成、句子分类等&…
阅读更多...
01.02、判定是否互为字符重排

01.02、判定是否互为字符重排

01.02、[简单] 判定是否互为字符重排 1、题目描述 给定两个由小写字母组成的字符串 s1 和 s2&#xff0c;请编写一个程序&#xff0c;确定其中一个字符串的字符重新排列后&#xff0c;能否变成另一个字符串。 在这道题中&#xff0c;我们的任务是判断两个字符串 s1 和 s2 是…
阅读更多...
《类和对象：基础原理全解析（上篇）》

《类和对象：基础原理全解析（上篇）》

目录 一、浅谈面向过程和面向对象二、C 中的结构体&#xff08;struct&#xff09;1. C 中 struct 的使用 三、C 中的类&#xff08;class&#xff09;四、类的封装性1. 类成员的权限控制关键字2. 权限控制关键字的使用 五、类的六大默认成员函数介绍六、构造函数1. 使用构造函…
阅读更多...
go基本知识与语法入门

go基本知识与语法入门

Go 语言&#xff08;又称 Golang&#xff09;是一种由 Google 开发的开源编程语言&#xff0c;设计目标是简洁、高效、并发友好&#xff0c;适合用于构建高性能的系统和网络应用程序。Go 语言的语法相对简单&#xff0c;非常适合大规模软件开发。 1. 基本结构 Go 程序的基本结…
阅读更多...
Jenkins 持续集成部署——Jenkins实战与运维（1）

Jenkins 持续集成部署——Jenkins实战与运维（1）

一、Jenkins 相关配置及代码发布 1. Jenkins 发布 php 代码 1.1 安装插件 先进入“系统管理”&#xff0c;再进入“管理插件”&#xff0c;在“已安装”中检查是否有“Git plugin”和“Publish Over SSH”两个插件&#xff0c;如果没有则需要安装&#xff0c;到“可选插件”中…
阅读更多...
HTTP 请求Media typetext/plain application/json text/json区别

HTTP 请求Media typetext/plain application/json text/json区别

这三种媒体类型表示的是内容在 HTTP 请求或响应中传输时的格式和语义&#xff0c;它们之间的主要区别如下&#xff1a; 1. text/plain 用途: 表示纯文本内容&#xff0c;没有格式化和结构化要求。 内容特征: 是简单的纯文本&#xff0c;没有特定的语法结构。 通常不包含…
阅读更多...
【网络云计算】2024第51周-每日【2024/12/19】小测-理论-如何实际一个校园网-简要列出

【网络云计算】2024第51周-每日【2024/12/19】小测-理论-如何实际一个校园网-简要列出

文章目录 1. 需求分析2. 网络架构3. 有线与无线网络覆盖4. 网络设备5. 安全策略6. 网络管理与监控7. 可扩展性与灵活性8. 教育应用与支持9. 用户教育与培训10. 预算与成本控制 【网络云计算】2024第51周-每日【2024/12/19】小测-理论-如何实际一个校园网 设计一个中专的校园网络…
阅读更多...
最新文章

Copyright @ 2022~2023