当前位置：首页 > news >正文

python异步协程

news 2026/3/30 17:24:39

`async`/`await` 和 `asyncio`

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

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())

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

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

异步上下文管理器

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

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())

事件循环执行阻塞操作 run_in_executor

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

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())

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

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

最佳实践是：

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

终止异步操作

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

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())

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

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

深入理解协程 async/await

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

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

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())

这种机制的优势在于：