pin_drop当前位置:知识文库 ❯ 图文
Python async/await语法完全指南：异步函数定义与调用 - 小确幸生活
一、概述
async/await 是 Python 中实现异步编程的核心语法。它们让异步代码的编写和阅读就像同步代码一样简单，是现代 Python 异步编程的基础。
async 用于定义协程函数，await 用于等待异步操作完成。这两个关键字配合使用，可以编写出高效、可读性强的异步代码。
二、语法
1. async def - 定义协程函数
代码示例
# 定义协程函数
async def my_coroutine():
"""这是一个协程函数"""
print("开始执行")
await asyncio.sleep(1) # 异步等待
print("执行完成")
return "结果"
# 协程函数调用返回协程对象
coro = my_coroutine() # 不会执行，只是创建协程对象
print(type(coro)) # <class 'coroutine'>2. await - 等待异步操作
代码示例
import asyncio
async def task1():
print("任务1开始")
await asyncio.sleep(2)
print("任务1完成")
return "任务1结果"
async def task2():
print("任务2开始")
await asyncio.sleep(1)
print("任务2完成")
return "任务2结果"
async def main():
# 使用 await 等待协程完成
result1 = await task1()
print(f"获得: {result1}")
result2 = await task2()
print(f"获得: {result2}")
# 运行协程
asyncio.run(main())3. async for - 异步迭代
代码示例
import asyncio
class AsyncCounter:
"""异步计数器"""
def __init__(self, limit):
self.limit = limit
self.current = 0
def __aiter__(self):
return self
async def __anext__(self):
if self.current >= self.limit:
raise StopAsyncIteration
self.current += 1
await asyncio.sleep(0.5) # 模拟异步操作
return self.current
async def main():
# 使用 async for 遍历异步迭代器
async for num in AsyncCounter(5):
print(f"数字: {num}")
asyncio.run(main())4. async with - 异步上下文管理
代码示例
import asyncio
class AsyncFileManager:
"""异步文件管理器"""
def __init__(self, filename):
self.filename = filename
async def __aenter__(self):
print(f"打开文件: {self.filename}")
await asyncio.sleep(0.5) # 模拟异步打开
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print(f"关闭文件: {self.filename}")
await asyncio.sleep(0.5) # 模拟异步关闭
async def write(self, data):
print(f"写入数据: {data}")
await asyncio.sleep(0.3)
async def main():
# 使用 async with 管理异步资源
async with AsyncFileManager("test.txt") as f:
await f.write("Hello, Async!")
await f.write("Second line")
asyncio.run(main())三、基本用法
1. 串行执行协程
代码示例
import asyncio
import time
async def fetch_data(url, delay):
"""模拟异步数据获取"""
print(f"开始获取: {url}")
await asyncio.sleep(delay)
print(f"完成获取: {url}")
return {"url": url, "data": f"数据_{url}"}
async def main():
start = time.time()
# 串行执行 - 一个接一个
result1 = await fetch_data("url1", 2)
result2 = await fetch_data("url2", 1)
result3 = await fetch_data("url3", 3)
print(f"\n总耗时: {time.time() - start:.2f} 秒") # 约 6 秒
print(f"结果: {result1}, {result2}, {result3}")
asyncio.run(main())2. 并发执行协程
代码示例
import asyncio
import time
async def fetch_data(url, delay):
print(f"开始获取: {url}")
await asyncio.sleep(delay)
print(f"完成获取: {url}")
return {"url": url, "data": f"数据_{url}"}
async def main():
start = time.time()
# 并发执行 - 同时开始
results = await asyncio.gather(
fetch_data("url1", 2),
fetch_data("url2", 1),
fetch_data("url3", 3)
)
print(f"\n总耗时: {time.time() - start:.2f} 秒") # 约 3 秒
print(f"结果: {results}")
asyncio.run(main())3. 嵌套协程调用
代码示例
import asyncio
async def inner_task(name, delay):
"""内部任务"""
print(f"[{name}] 开始")
await asyncio.sleep(delay)
print(f"[{name}] 完成")
return f"{name}_result"
async def outer_task():
"""外部任务 - 调用内部任务"""
print("外部任务开始")
# 调用其他协程
result1 = await inner_task("任务A", 1)
result2 = await inner_task("任务B", 2)
print(f"外部任务完成，获得: {result1}, {result2}")
return f"外部结果: {result1}, {result2}"
async def main():
result = await outer_task()
print(f"最终结果: {result}")
asyncio.run(main())提示：协程可以嵌套调用，一个协程可以 await 其他协程，形成调用链。
四、代码示例
示例 1：异步装饰器
代码示例
import asyncio
import time
from functools import wraps
def async_timer(func):
"""异步计时装饰器"""
@wraps(func)
async def wrapper(*args, **kwargs):
start = time.time()
result = await func(*args, **kwargs)
elapsed = time.time() - start
print(f"{func.__name__} 耗时: {elapsed:.2f} 秒")
return result
return wrapper
@async_timer
async def slow_operation(name, delay):
"""慢速操作"""
print(f"[{name}] 开始")
await asyncio.sleep(delay)
print(f"[{name}] 完成")
return f"{name}_done"
async def main():
result = await slow_operation("任务X", 2)
print(f"结果: {result}")
asyncio.run(main())示例 2：异步错误处理
代码示例
import asyncio
async def risky_task(name, should_fail=False):
"""可能失败的任务"""
print(f"[{name}] 开始")
await asyncio.sleep(1)
if should_fail:
raise ValueError(f"{name} 执行失败")
print(f"[{name}] 成功")
return f"{name}_result"
async def main():
# 方式1: try-except 捕获单个协程异常
try:
result = await risky_task("任务1", should_fail=True)
except ValueError as e:
print(f"捕获异常: {e}")
# 方式2: gather 收集多个协程的异常
results = await asyncio.gather(
risky_task("任务2", should_fail=False),
risky_task("任务3", should_fail=True),
risky_task("任务4", should_fail=False),
return_exceptions=True # 返回异常而不是抛出
)
print("\n所有结果:")
for i, result in enumerate(results, 2):
if isinstance(result, Exception):
print(f" 任务{i}: 异常 - {result}")
else:
print(f" 任务{i}: 成功 - {result}")
asyncio.run(main())示例 3：异步生成器
代码示例
import asyncio
async def async_range(start, end, delay=0.5):
"""异步范围生成器"""
for i in range(start, end):
await asyncio.sleep(delay) # 模拟异步操作
yield i
async def main():
# 使用 async for 遍历异步生成器
print("异步生成器示例:")
async for num in async_range(1, 6, 0.3):
print(f"获得: {num}")
asyncio.run(main())示例 4：异步上下文管理器实战
代码示例
import asyncio
class AsyncDatabaseConnection:
"""异步数据库连接"""
def __init__(self, db_name):
self.db_name = db_name
self.connected = False
async def __aenter__(self):
print(f"连接到数据库: {self.db_name}")
await asyncio.sleep(1) # 模拟连接延迟
self.connected = True
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
print(f"断开数据库连接: {self.db_name}")
await asyncio.sleep(0.5) # 模拟断开延迟
self.connected = False
async def query(self, sql):
if not self.connected:
raise RuntimeError("未连接到数据库")
print(f"执行查询: {sql}")
await asyncio.sleep(0.8) # 模拟查询延迟
return f"查询结果: {sql}"
async def main():
async with AsyncDatabaseConnection("mydb") as db:
result1 = await db.query("SELECT * FROM users")
print(result1)
result2 = await db.query("SELECT * FROM orders")
print(result2)
print(f"连接状态: {db.connected}") # False
asyncio.run(main())五、注意事项
⚠️ 注意 1：必须在协程中使用 await
await只能在async def定义的协程函数中使用。在普通函数中使用 await 会导致语法错误。
⚠️ 注意 2：不要忘记 await
调用协程函数时如果忘记使用 await，协程不会执行，只会返回一个协程对象。这是初学者常见的错误。
⚠️ 注意 3：避免阻塞操作
在协程中避免使用阻塞操作（如 time.sleep()、同步 I/O），这会阻塞整个事件循环。应使用异步版本（asyncio.sleep()、aiohttp 等）。
⚠️ 注意 4：正确处理异常
使用 try-except 捕获协程中的异常。使用 asyncio.gather() 时，可以设置 return_exceptions=True 来收集异常而不是立即抛出。
六、小结
-
async def：定义协程函数，创建可等待的异步函数
-
await：等待异步操作完成，只能在协程中使用
-
async for：异步迭代，遍历异步迭代器
-
async with：异步上下文管理，管理异步资源
-
注意事项：必须在协程中使用 await、避免阻塞操作、正确处理异常
七、练习题
练习 1
编写一个异步程序，使用 async/await 语法实现以下功能：
1. 定义一个协程函数 fetch_data(url, delay)，模拟异步数据获取
2. 创建 3 个任务，分别获取不同 URL 的数据（延迟时间不同）
3. 使用 asyncio.gather() 并发执行所有任务
4. 打印所有结果和总耗时
练习 2
实现一个异步上下文管理器 AsyncTimer，用于测量代码块的执行时间：
1. 在 __aenter__ 中记录开始时间
2. 在 __aexit__ 中计算并打印执行时间
3. 使用 async with 语法测试该上下文管理器
4. 在代码块中执行一些异步操作（如 asyncio.sleep()）
常见问题
async def 和普通 def 有什么区别？
async def 定义的是协程函数，调用时返回协程对象，需要使用 await 或 asyncio.run() 来执行。普通 def 定义的是同步函数，调用时立即执行并返回结果。协程函数可以在执行过程中暂停和恢复，实现异步操作。
为什么调用协程函数必须使用 await？
调用协程函数只会创建协程对象，不会执行协程。必须使用 await 来等待协程执行完成并获取结果。如果忘记使用 await，协程不会执行，这是初学者常见的错误。可以使用 asyncio.run() 在顶层执行协程。
await 可以在普通函数中使用吗？
不可以。await 只能在 async def 定义的协程函数中使用。在普通函数中使用 await 会导致 SyntaxError。如果需要在普通函数中调用协程，可以使用 asyncio.run() 或 loop.run_until_complete()。
如何在协程中处理异常？
使用 try-except 块捕获协程中的异常。对于 asyncio.gather()，可以设置 return_exceptions=True 参数，这样会返回异常对象而不是抛出。对于单个协程，使用 asyncio.wait_for() 可以捕获超时异常。
本文涉及AI创作
内容由AI创作，请仔细甄别
