Python OOP 设计思想 18:异步接口与协作

# 同步接口：立即得到结果，阻塞调用者def fetch_data_sync() -> Data:    """同步获取数据：立即返回结果，阻塞当前线程"""    return get_data_from_source()    # 阻塞直到完成# 异步接口：返回可等待对象，支持协作async def fetch_data_async() -> Data:    """异步获取数据：返回协程，可等待完成"""    return await get_data_async()    # 挂起并让出控制权# 使用对比data_sync = fetch_data_sync()           # 阻塞调用线程，无法并发data_async = await fetch_data_async()   # 挂起当前协程，允许其他任务运行

async def download_file(url: str) -> bytes:    """异步下载文件：必须使用 await 调用"""    async with aiohttp.ClientSession() as session:        async with session.get(url) as response:            return await response.read()# 明确的调用方式content = await download_file("http://example.com/data.txt")  # ✅ 正确的异步调用

# 错误的异步函数：内部阻塞async def bad_async_operation():    time.sleep(1)             # 阻塞整个事件循环    # 所有其他协程都会被阻塞1秒# 正确的异步函数：协作式等待async def good_async_operation():    await asyncio.sleep(1)    # 让出控制权，允许其他协程运行    # 其他协程可以在此期间执行

async def process_large_file(file_path: str):    """处理大文件：分块处理并定期让出控制权    注意：示例中使用同步文件读取，重点在于协作式让出控制权的语义    """    chunk_size = 1024 * 1024  # 1MB    with open(file_path, 'rb') as f:        while chunk := f.read(chunk_size):            # 处理当前块            process_chunk(chunk)            # 主动让出控制权，保持系统响应性            await asyncio.sleep(0)  # 允许其他任务运行

data = await reader.read_async()

from typing import Awaitable, Anyfrom abc import ABC, abstractmethodclass AsyncReadable(ABC):    """异步可读接口抽象"""    @abstractmethod    async def read(self) -> str:        """异步读取数据"""        passclass FileReader(AsyncReadable):    """文件读取实现"""    async def read(self) -> str:        return await self._read_file()    async def _read_file(self) -> str:        # 实际的异步文件读取逻辑        await asyncio.sleep(0.1)        return "file content"class NetworkReader(AsyncReadable):    """网络读取实现"""    async def read(self) -> str:        return await self._fetch_from_network()    async def _fetch_from_network(self) -> str:        # 实际的异步网络请求逻辑        await asyncio.sleep(0.2)        return "network data"# 统一的使用方式async def use_reader(reader: AsyncReadable):    """接受任何实现 AsyncReadable 接口的对象"""    content = await reader.read()  # ✅ 统一异步接口    print(f"Read: {content}")

import aiosqlitefrom contextlib import asynccontextmanagerclass DatabaseConnection:    """数据库连接：异步上下文管理"""    async def __aenter__(self):        """进入上下文：建立连接"""        self.conn = await aiosqlite.connect("app.db")        await self._initialize()        return self    async def __aexit__(self, exc_type, exc_val, exc_tb):        """退出上下文：清理资源"""        try:            if exc_type is not None:                await self.conn.rollback()  # 异常时回滚            else:                await self.conn.commit()    # 正常时提交        finally:            await self.conn.close()         # 总是关闭连接    async def _initialize(self):        """初始化连接设置"""        await self.conn.execute("PRAGMA foreign_keys = ON")        await self.conn.execute("PRAGMA journal_mode = WAL")    async def execute_query(self, sql: str, params=None):        """执行查询"""        async with self.conn.execute(sql, params or ()) as cursor:            return await cursor.fetchall()# 使用异步上下文管理器async def process_user_data(user_id: int):    """使用数据库连接处理用户数据"""    async with DatabaseConnection() as db:  # ✅ 自动管理生命周期        # 在此块内，连接是活跃的        results = await db.execute_query(            "SELECT * FROM users WHERE id = ?",             (user_id,)        )        # 退出时自动提交和关闭    return results

from abc import ABC, abstractmethodfrom typing import Dict, Optionalclass UserRepository(ABC):    """用户存储库抽象接口"""    @abstractmethod    async def get_user(self, user_id: int) -> Optional[Dict]:        """根据ID获取用户"""        pass    @abstractmethod    async def save_user(self, user: Dict) -> bool:        """保存用户"""        pass    @abstractmethod    async def delete_user(self, user_id: int) -> bool:        """删除用户"""        passclass InMemoryUserRepository(UserRepository):    """内存实现：用于测试"""    def __init__(self):        self._users = {}    async def get_user(self, user_id: int) -> Optional[Dict]:        await asyncio.sleep(0.01)  # 模拟轻微延迟        return self._users.get(user_id)    async def save_user(self, user: Dict) -> bool:        await asyncio.sleep(0.01)        self._users[user["id"]] = user        return Trueclass DatabaseUserRepository(UserRepository):    """数据库实现：用于生产"""    def __init__(self, db_pool):        self.db_pool = db_pool    async def get_user(self, user_id: int) -> Optional[Dict]:        async with self.db_pool.acquire() as conn:            async with conn.execute(                "SELECT * FROM users WHERE id = ?",                 (user_id,)            ) as cursor:                row = await cursor.fetchone()                return dict(row) if row else None# 统一的使用方式async def process_user(repo: UserRepository, user_id: int):    """接受任何用户存储库实现"""    user = await repo.get_user(user_id)  # ✅ 统一接口调用    if user:        # 处理用户        return process(user)    return None

import asyncioclass UserService:    """用户服务：组合多个异步操作"""    def __init__(self, user_repo: UserRepository,                  profile_repo: 'ProfileRepository'):        self.user_repo = user_repo        self.profile_repo = profile_repo    async def get_user_with_profile(self, user_id: int) -> Dict:        """并行获取用户基本信息和详情"""        # 并行执行多个异步操作        user_task = asyncio.create_task(self.user_repo.get_user(user_id))        profile_task = asyncio.create_task(            self.profile_repo.get_profile(user_id)        )        # 等待所有任务完成        user, profile = await asyncio.gather(            user_task,             profile_task,            return_exceptions=False  # 任何异常都会传播        )        # 组合结果        if user and profile:            return {**user, "profile": profile}        return None    async def batch_process_users(self, user_ids: list[int]):        """批量处理用户：使用异步生成器"""        # 创建所有任务        tasks = [self.process_single_user(uid) for uid in user_ids]        # 按完成顺序处理结果        for completed in asyncio.as_completed(tasks):            result = await completed            if result:                yield result    async def process_single_user(self, user_id: int):        """处理单个用户（支持取消）"""        try:            return await asyncio.wait_for(                self.user_repo.get_user(user_id),                timeout=5.0  # 5秒超时            )        except asyncio.TimeoutError:            print(f"获取用户 {user_id} 超时")            return None

from typing import Awaitableasync def download_file_with_progress(    url: str,    destination: str,    chunk_size: int = 8192,    progress_callback = None) -> None:    """    异步下载文件并显示进度    Args:        url: 文件URL地址        destination: 本地保存路径        chunk_size: 每次读取的字节大小        progress_callback: 进度回调函数，接收 (downloaded, total) 参数    Returns:        None: 文件下载完成后无返回值    Raises:        aiohttp.ClientError: 网络请求失败时        IOError: 文件写入失败时    Notes:        - 必须使用 await 调用        - 支持取消操作，取消时会清理临时文件        - 进度回调在事件循环中调用，不应执行阻塞操作    """    # 实现细节...    pass# 清晰的调用方式try:    await download_file_with_progress(        url="http://example.com/largefile.zip",        destination="/data/largefile.zip",        progress_callback=lambda d, t: print(f"进度: {d/t:.1%}")    )except aiohttp.ClientError as e:    print(f"下载失败: {e}")

from typing import AsyncIterable, AsyncIterator, Optionalfrom dataclasses import dataclass@dataclassclass DownloadResult:    """下载结果数据类"""    success: bool    bytes_downloaded: int    duration: float    error: Optional[Exception] = Noneasync def download_with_retry(    url: str,    max_retries: int = 3) -> AsyncIterator[DownloadResult]:    """    带重试的下载器    Returns:        AsyncIterator[DownloadResult]: 异步迭代器，每次迭代返回进度    Example:        >>> async for result in download_with_retry(url):        ...     print(f"进度: {result.bytes_downloaded} bytes")    """    for attempt in range(max_retries):        try:            # 尝试下载            yield DownloadResult(                success=False,  # 还在进行中                bytes_downloaded=0,                duration=0.0            )            # ... 实际下载逻辑            break  # 成功则退出重试循环        except Exception as e:            if attempt == max_retries - 1:                yield DownloadResult(                    success=False,                    bytes_downloaded=0,                    duration=0.0,                    error=e                )

import asynciofrom typing import TypeVar, CallableT = TypeVar('T')async def with_timeout(    coro: Awaitable[T],    timeout: float,    default: T = None) -> T:    """    带超时的异步操作    Args:        coro: 要执行的协程        timeout: 超时时间（秒）        default: 超时时的默认返回值    Returns:        协程结果或默认值    Note:        超时会取消被等待的协程；如需保留任务，应使用 asyncio.shield    """    try:        return await asyncio.wait_for(coro, timeout=timeout)    except asyncio.TimeoutError:        return defaultasync def with_retry(    coro_func: Callable[[], Awaitable[T]],    max_retries: int = 3,    delay: float = 1.0,    backoff_factor: float = 2.0) -> T:    """    带指数退避的重试机制    Args:        coro_func: 返回协程的函数（每次重试重新创建协程）        max_retries: 最大重试次数        delay: 初始延迟（秒）        backoff_factor: 退避因子    Returns:        最终结果    Raises:        最后一次尝试的异常    """    last_exception = None    for attempt in range(max_retries):        try:            return await coro_func()        except Exception as e:            last_exception = e            # 最后一次尝试，直接抛出异常            if attempt == max_retries - 1:                raise            # 计算退避延迟            wait_time = delay * (backoff_factor ** attempt)            print(f"尝试 {attempt + 1} 失败，{wait_time:.1f}秒后重试")            await asyncio.sleep(wait_time)    # 理论上不会执行到这里    raise last_exceptionclass ResilientService:    """具有弹性的异步服务"""    def __init__(self):        self._tasks = set()    async def resilient_operation(self, operation: Awaitable[T]) -> T:        """弹性操作：结合超时和重试"""        async def attempt():            return await with_timeout(operation, timeout=10.0)        return await with_retry(            attempt,            max_retries=3,            delay=2.0        )    async def safe_background_task(self, coro: Awaitable):        """安全的后台任务：自动捕获和记录异常"""        task = asyncio.create_task(coro)        self._tasks.add(task)        def cleanup(t):            self._tasks.discard(t)            if t.exception():                # 记录异常但不传播                print(f"后台任务异常: {t.exception()}")        task.add_done_callback(cleanup)        return task

class CancellableOperation:    """明确支持取消的异步操作"""    def __init__(self):        self._cancelled = False        self._cancel_event = asyncio.Event()    async def run(self):        """运行操作，定期检查取消状态"""        for i in range(10):            if self._cancelled:                self._cleanup()  # 执行清理                raise asyncio.CancelledError("操作被取消")            # 执行一步操作            await self._step(i)            # 定期检查取消事件            try:                await asyncio.wait_for(                    self._cancel_event.wait(),                    timeout=0.1  # 每0.1秒检查一次                )                self._cancelled = True            except asyncio.TimeoutError:                pass  # 未取消，继续执行    def cancel(self):        """请求取消操作"""        self._cancelled = True        self._cancel_event.set()    async def _step(self, iteration: int):        """单个步骤的实现"""        await asyncio.sleep(0.5)  # 模拟工作        print(f"步骤 {iteration} 完成")    def _cleanup(self):        """取消时的清理操作"""        print("执行取消清理...")