Python24_async with语法

Python24_async with 语法

文章目录

• Python24_async with 语法
• • @[toc]
  • 1. 基础概念
  • • 1.1 什么是 `async with`？
    • 1.2 为什么需要 `async with`？
  • 2. 核心原理
  • • 2.1 异步上下文管理器协议
    • 2.2 执行流程
  • 3. 常见使用场景
  • • 3.1 异步文件操作（aiofiles）
    • 3.2 异步数据库连接（aiomysql/aiopg）
    • 3.3 异步 HTTP 请求（aiohttp）
    • 3.4 异步锁（asyncio.Lock）
  • 4. 常见问题与解答（Q&A）
  • • Q1: `async with` 和 `with` 有什么区别？
    • Q2: 如何自定义异步上下文管理器？
    • Q3: `async with` 可以嵌套使用吗？
    • Q4: `__aexit__` 的参数是什么意思？
    • Q5: 可以在 `async with` 中使用 `await` 吗？
    • Q6: 如何正确处理异常？
    • Q7: 异步上下文管理器在 `__aexit__` 中可以再使用 `await` 吗？
  • 5. 最佳实践
  • • 5.1 使用 `@asynccontextmanager` 简化代码
    • 5.2 超时控制
    • 5.3 避免常见错误
  • 6. 速查表
  • 7. 相关知识点延伸

1. 基础概念

1.1 什么是async with？

async with是 Python 异步上下文管理器（Asynchronous Context Manager）的语法糖，用于在异步代码中管理资源的获取和释放。

asyncwith表达式as变量:# 异步代码块pass

1.2 为什么需要async with？

2. 核心原理

2.1 异步上下文管理器协议

一个对象要成为异步上下文管理器，必须实现以下两个异步方法：

classAsyncContextManager:asyncdef__aenter__(self):"""进入上下文时调用"""returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):"""退出上下文时调用"""# 处理异常或清理资源pass

2.2 执行流程

┌─────────────────┐ │ async with │ │ 表达式 as x: │ └────────┬────────┘ ▼ ┌─────────────────┐ │ await __aenter__() │ ← 异步获取资源 │ 返回值赋给 x │ └────────┬────────┘ ▼ ┌─────────────────┐ │ 执行代码块 │ └────────┬────────┘ ▼ ┌─────────────────┐ │ await __aexit__() │ ← 异步释放资源 │ (无论是否异常都执行) │ └─────────────────┘

3. 常见使用场景

3.1 异步文件操作（aiofiles）

importaiofilesimportasyncioasyncdefread_file():asyncwithaiofiles.open('data.txt','r')asf:content=awaitf.read()print(content)asyncio.run(read_file())

3.2 异步数据库连接（aiomysql/aiopg）

importaiomysqlasyncdefquery_db():asyncwithaiomysql.create_pool(host='localhost',user='root',password='pwd',db='test')aspool:asyncwithpool.acquire()asconn:asyncwithconn.cursor()ascur:awaitcur.execute("SELECT * FROM users")result=awaitcur.fetchall()returnresult

3.3 异步 HTTP 请求（aiohttp）

importaiohttpimportasyncioasyncdeffetch_url(url):asyncwithaiohttp.ClientSession()assession:asyncwithsession.get(url)asresponse:returnawaitresponse.text()# 使用html=asyncio.run(fetch_url('https://api.github.com'))

3.4 异步锁（asyncio.Lock）

importasyncio lock=asyncio.Lock()counter=0asyncdefincrement():globalcounterasyncwithlock:# 确保同一时间只有一个协程修改 countercurrent=counterawaitasyncio.sleep(0.1)# 模拟耗时操作counter=current+1asyncdefmain():awaitasyncio.gather(*[increment()for_inrange(10)])print(f"Counter:{counter}")# 输出: Counter: 10asyncio.run(main())

4. 常见问题与解答（Q&A）

Q1:async with和with有什么区别？

答：

错误示例：

# ❌ 错误：在异步函数中使用同步 with 管理异步资源asyncdefwrong():withaiohttp.ClientSession()assession:# 错误！pass

Q2: 如何自定义异步上下文管理器？

答：

方式一：类实现

classAsyncDatabase:asyncdef__aenter__(self):self.conn=awaitcreate_connection()returnself.connasyncdef__aexit__(self,exc_type,exc,tb):awaitself.conn.close()# 返回 True 表示异常已处理，不再传播returnFalse# 使用asyncwithAsyncDatabase()asconn:awaitconn.query("SELECT 1")

方式二：装饰器（asynccontextmanager）

fromcontextlibimportasynccontextmanager@asynccontextmanagerasyncdefmanaged_resource():print("获取资源...")resource=awaitcreate_async_resource()try:yieldresourcefinally:print("释放资源...")awaitresource.cleanup()# 使用asyncwithmanaged_resource()asres:awaitres.do_something()

Q3:async with可以嵌套使用吗？

答：可以，支持多种写法：

# 方式一：嵌套asyncwithA()asa:asyncwithB()asb:pass# 方式二：单行（Python 3.10+）asyncwithA()asa,B()asb:pass# 方式三：括号（Python 3.10+，推荐）asyncwith(A()asa,B()asb,C()asc):pass

Q4:__aexit__的参数是什么意思？

答：

asyncdef__aexit__(self,exc_type,exc_val,exc_tb):""" exc_type: 异常类型（如 ValueError） exc_val: 异常实例 exc_tb: 异常追踪信息 """ifexc_typeisnotNone:print(f"发生异常:{exc_val}")# 返回 True 会抑制异常，False 或 None 会传播异常returnTrue

示例：

classSuppressError:asyncdef__aenter__(self):returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):ifexc_type==ValueError:print(f"捕获 ValueError:{exc_val}")returnTrue# 抑制异常returnFalse# 其他异常正常抛出asyncdeftest():asyncwithSuppressError():raiseValueError("测试错误")# 被捕获，不会抛出print("继续执行")

Q5: 可以在async with中使用await吗？

答：可以，而且这是常态：

asyncdefprocess():asyncwithget_session()assession:result=awaitsession.fetch_data()# ✅ 正常awaitsession.save(result)# ✅ 正常

Q6: 如何正确处理异常？

答：

classSafeAsyncContext:asyncdef__aenter__(self):returnselfasyncdef__aexit__(self,exc_type,exc_val,exc_tb):# 清理资源（无论是否异常都会执行）awaitself.cleanup()# 选择是否处理异常ifexc_typeisasyncio.CancelledError:returnTrue# 抑制取消异常returnFalse# 其他异常继续传播# 外部捕获asyncdefmain():try:asyncwithSafeAsyncContext():raiseValueError("测试")exceptValueError:print("捕获到 ValueError")

Q7: 异步上下文管理器在__aexit__中可以再使用await吗？

答：可以，这是设计用途之一：

classAsyncConnection:asyncdef__aexit__(self,exc_type,exc_val,exc_tb):# 异步关闭连接awaitself.conn.close()# 异步记录日志awaitlog_async("连接已关闭")# 异步发送指标awaitmetrics.send("connection_closed")

5. 最佳实践

5.1 使用@asynccontextmanager简化代码

fromcontextlibimportasynccontextmanager@asynccontextmanagerasyncdeftransaction(db):awaitdb.begin()try:yielddbawaitdb.commit()exceptException:awaitdb.rollback()raise# 使用asyncwithtransaction(database)asdb:awaitdb.execute("INSERT ...")

5.2 超时控制

importasyncioasyncdefwith_timeout():try:asyncwithasyncio.timeout(5):# Python 3.11+asyncwithaiohttp.ClientSession()assession:asyncwithsession.get('https://slow.com')asresp:returnawaitresp.text()exceptasyncio.TimeoutError:print("请求超时")

5.3 避免常见错误

# ❌ 错误：忘记 awaitasyncwithsome_async_context():# 正确passwithsome_async_context():# 错误！协程没有被 await# ❌ 错误：在同步函数中使用 async withdefsync_function():asyncwithsomething():# 错误！同步函数不能有 awaitpass# ✅ 正确：使用 asyncio.run 或 awaitasyncdefasync_function():asyncwithsomething():pass

6. 速查表

7. 相关知识点延伸

• async for- 异步迭代器
• asyncio.gather()- 并发执行多个协程
• asyncio.create_task()- 创建后台任务
• contextlib.AsyncExitStack- 动态管理多个异步上下文