本文目录导读：
在Python异步编程中,保障异步安全主要涉及以下几个方面，我将通过具体案例来说明：
共享状态保护（线程安全 vs 协程安全）
错误案例
import asyncio
# 错误的共享状态
shared_data = [] # 没有保护
async def writer():
for i in range(100):
shared_data.append(i)
await asyncio.sleep(0.01)
async def reader():
while True:
if len(shared_data) > 50:
print(f"数据长度: {len(shared_data)}")
break
await asyncio.sleep(0.01)
async def main():
await asyncio.gather(writer(), reader())
asyncio.run(main())
安全案例（使用锁）
import asyncio
from asyncio import Lock
class SafeSharedData:
def __init__(self):
self._data = []
self._lock = Lock()
async def add(self, item):
async with self._lock:
self._data.append(item)
async def get_length(self):
async with self._lock:
return len(self._data)
async def get_snapshot(self):
async with self._lock:
return self._data.copy()
async def writer(data: SafeSharedData):
for i in range(100):
await data.add(i)
await asyncio.sleep(0.01)
async def reader(data: SafeSharedData):
while True:
length = await data.get_length()
if length > 50:
snapshot = await data.get_snapshot()
print(f"数据长度: {length}")
break
await asyncio.sleep(0.01)
async def main():
data = SafeSharedData()
await asyncio.gather(writer(data), reader(data))
asyncio.run(main())
数据库连接池安全
安全案例
import asyncio
from typing import Optional
from contextlib import asynccontextmanager
class AsyncConnectionPool:
def __init__(self, max_connections=5):
self.max_connections = max_connections
self._connections = []
self._semaphore = asyncio.Semaphore(max_connections)
self._lock = asyncio.Lock()
@asynccontextmanager
async def get_connection(self):
async with self._semaphore:
conn = await self._create_connection()
try:
yield conn
finally:
await self._release_connection(conn)
async def _create_connection(self) -> dict:
# 模拟创建连接
conn = {"id": id(self), "state": "active"}
async with self._lock:
self._connections.append(conn)
return conn
async def _release_connection(self, conn):
async with self._lock:
if conn in self._connections:
self._connections.remove(conn)
conn["state"] = "released"
# 使用示例
async def query_database(pool: AsyncConnectionPool, query_id: int):
async with pool.get_connection() as conn:
# 模拟数据库查询
await asyncio.sleep(0.1)
print(f"查询 {query_id} 完成，连接状态: {conn['state']}")
return f"结果_{query_id}"
async def main():
pool = AsyncConnectionPool(max_connections=3)
tasks = [query_database(pool, i) for i in range(10)]
results = await asyncio.gather(*tasks, return_exceptions=True)
print(f"结果: {results}")
asyncio.run(main())
异步上下文管理器安全
安全案例
import asyncio
from contextlib import asynccontextmanager
class ResourceManager:
def __init__(self):
self._resource = None
self._in_use = False
async def acquire(self):
if self._in_use:
raise RuntimeError("资源已被占用")
self._resource = {"status": "acquired"}
self._in_use = True
return self._resource
async def release(self):
if self._resource:
self._resource["status"] = "released"
self._in_use = False
@asynccontextmanager
async def use_resource(self):
resource = await self.acquire()
try:
yield resource
except Exception as e:
print(f"使用资源时出错: {e}")
raise
finally:
await self.release()
# 安全使用
async def process_data(manager: ResourceManager):
async with manager.use_resource() as resource:
# 处理数据
await asyncio.sleep(0.1)
resource["data"] = "processed"
print(f"资源状态: {resource}")
# 即使在处理中出错，资源也会被释放
# 不安全使用示例（不推荐）
async def unsafe_process(manager: ResourceManager):
# 手动管理资源容易出错
resource = await manager.acquire()
try:
await asyncio.sleep(0.1)
# 可能忘记释放资源
finally:
await manager.release()
异步任务取消安全
import asyncio
from typing import Set
class TaskManager:
def __init__(self):
self._tasks: Set[asyncio.Task] = set()
self._lock = asyncio.Lock()
async def add_task(self, coro):
task = asyncio.create_task(self._safe_run(coro))
self._tasks.add(task)
task.add_done_callback(self._tasks.discard)
return task
async def _safe_run(self, coro):
try:
return await coro
except asyncio.CancelledError:
# 捕获取消信号，进行清理
print("任务被取消，进行清理")
raise
except Exception as e:
print(f"任务出错: {e}")
raise
async def cancel_all(self):
async with self._lock:
for task in self._tasks:
task.cancel()
# 等待所有任务完成取消
await asyncio.gather(*self._tasks, return_exceptions=True)
self._tasks.clear()
@property
def active_count(self):
return len(self._tasks)
# 使用案例
async def long_running_task(task_id: int):
try:
print(f"任务 {task_id} 开始")
await asyncio.sleep(5)
print(f"任务 {task_id} 完成")
return task_id
except asyncio.CancelledError:
print(f"任务 {task_id} 被取消")
raise
async def main():
manager = TaskManager()
# 添加任务
tasks = [manager.add_task(long_running_task(i)) for i in range(3)]
# 等待1秒后取消所有任务
await asyncio.sleep(1)
await manager.cancel_all()
print(f"活动任务数: {manager.active_count}")
asyncio.run(main())
异步文件操作安全
import asyncio
import aiofiles
from pathlib import Path
class SafeFileHandler:
def __init__(self, file_path: str):
self.file_path = Path(file_path)
self._lock = asyncio.Lock()
self._file = None
self._is_open = False
async def __aenter__(self):
await self._open()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
await self._close()
async def _open(self):
if not self._is_open:
self._file = await aiofiles.open(self.file_path, 'a+')
self._is_open = True
async def _close(self):
if self._is_open and self._file:
await self._file.close()
self._is_open = False
self._file = None
async def write_safe(self, content: str):
async with self._lock:
await self._file.write(content + '\n')
await self._file.flush()
async def read_safe(self) -> str:
async with self._lock:
await self._file.seek(0)
return await self._file.read()
# 使用示例
async def log_writer(handler: SafeFileHandler, task_id: int):
for i in range(5):
await handler.write_safe(f"任务 {task_id}: 日志 {i}")
await asyncio.sleep(0.1)
async def main():
async with SafeFileHandler("safe_log.txt") as handler:
tasks = [log_writer(handler, i) for i in range(3)]
await asyncio.gather(*tasks)
# 读取所有内容
content = await handler.read_safe()
print("日志内容:")
print(content)
asyncio.run(main())
关键安全原则总结
- 使用锁机制：对共享状态使用
asyncio.Lock - 使用信号量：控制并发数量用
asyncio.Semaphore - 使用上下文管理器：确保资源正确释放
- 任务取消处理：正确捕获
CancelledError - 避免阻塞操作：使用异步版本的 I/O 操作
- 最小化锁范围：只在必要时加锁
- 处理异常：在任何可能出错的异步操作中处理异常
这些案例展示了如何在Python异步编程中保障执行安全,避免竞态条件和资源泄漏等问题。
