Python异步安全案例如何保障异步执行

wen python案例 2026-06-28 273

本文目录导读：

共享状态保护（线程安全 vs 协程安全）
数据库连接池安全
异步上下文管理器安全
异步任务取消安全
异步文件操作安全
关键安全原则总结

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