SuperAPI AI 工具集
合作ENAI 行业资讯收藏
Skill 广场async-python-patterns

async-python-patterns

编程与开发

查看项目

更新于 10-15 09:51

技能介绍

用途

Async Python Patterns 用于指导开发者使用 asyncio、并发编程模式与 async/await 构建高性能、非阻塞的异步 Python 系统。其触发场景包括：构建异步 Web API（FastAPI、aiohttp、Sanic）、实现并发 I/O 操作（数据库、文件、网络）、创建并发请求的网页抓取器、开发实时应用（WebSocket 服务器、聊天系统）、同时处理多个独立任务、构建异步通信的微服务、优化 I/O 密集型工作负载、实现异步后台任务与队列，以及在 threading、multiprocessing 与 asyncio 之间做选择。

核心概念

  • 事件循环：单线程协作式多任务，调度协程执行，以非阻塞方式处理 I/O，管理回调与 Future。
  • 协程：用 async def 定义、可暂停与恢复的函数。
  • 任务：在事件循环上并发运行的已调度协程。
  • Future：表示异步操作最终结果的底层对象。
  • 异步上下文管理器：支持 async with 以正确清理资源。
  • 异步迭代器：支持 async for 以遍历异步数据源。

快速开始示例：

import asyncio

async def main():
    print("Hello")
    await asyncio.sleep(1)
    print("World")

asyncio.run(main())

主要能力（基础模式）

  • 基础 async/await：用 await 调用协程并获取结果。
  • 使用 gather() 并发执行：将多个协程任务收集后并发运行。
  • 任务创建与管理：用 asyncio.create_task 创建后台任务，在等待期间执行其他工作。
  • 错误处理：在协程内 try/except，并配合 gather(..., return_exceptions=True) 区分成功与失败结果。
  • 超时处理：使用 asyncio.wait_for 设置超时并捕获 asyncio.TimeoutError。
  • Semaphore 限流：用 asyncio.Semaphore 限制最大并发数。

使用方式

技能以模式与代码示例的形式呈现，涵盖从入口点 asyncio.run()、协程等待、并发收集、任务管理，到错误处理、超时与限流的具体写法。同时提供决策框架与常见陷阱说明，帮助开发者在实际编码中正确组织异步代码。

适用场景

  • I/O 密集型任务（网络、数据库、文件）。
  • 大量并发连接。
  • Web 服务器与 API 客户端。
  • WebSocket 应用。
  • 并发网页抓取、实时应用、异步微服务与后台任务队列。

决策框架

  • 使用 asyncio：I/O 密集型任务、大量并发连接、Web 服务器与 API 客户端、WebSocket 应用。
  • 使用 threading：I/O 密集型且依赖现有同步库、简单并行需求、与释放 GIL 的 C 扩展交互。
  • 使用 multiprocessing：CPU 密集型任务（计算、数据处理）、需要真正并行（绕过 GIL）、不需要共享内存的任务。

常见陷阱

  • 忘记 await：直接调用异步函数会返回协程对象，必须 await 才会执行。
  • 阻塞事件循环：在异步函数中使用 time.sleep 会阻塞，应改用 await asyncio.sleep。
  • 未处理取消：应捕获 asyncio.CancelledError 执行清理后重新抛出以传播。
  • 混用同步与异步代码：同步函数中不能直接 await，应使用 asyncio.run 等方式调用。

测试

使用 pytest 与 pytest-asyncio，通过 @pytest.mark.asyncio 标记异步测试函数，可断言结果，也可用 pytest.raises(asyncio.TimeoutError) 验证超时行为。

最佳实践

  • 使用 asyncio.run() 作为入口点（Python 3.7+）。
  • 始终 await 协程以执行它们。
  • 使用 gather() 并发执行多个任务。
  • 用 try/except 实现正确的错误处理。
  • 使用超时防止操作挂起。
  • 连接池以提升性能。
  • 避免在异步代码中执行阻塞操作，改用 run_in_executor。
  • 使用信号量进行限流。
  • 正确处理任务取消。
  • 使用 pytest-asyncio 测试异步代码。

参考资料与资源

  • references/async-patterns.md：异步上下文管理器、异步迭代器/生成器、生产者-消费者模式、异步锁与同步、使用 aiohttp 的网页抓取、异步数据库操作、WebSocket 服务器实现、连接池、批量操作、在 executor 中运行阻塞操作。
  • Python asyncio 文档：https://docs.python.org/3/library/asyncio.html
  • aiohttp：异步 HTTP 客户端/服务器。
  • FastAPI：现代异步 Web 框架。
  • asyncpg：异步 PostgreSQL 驱动。
  • motor：异步 MongoDB 驱动。

限制

资料未提供该技能的安装命令、收费方式、价格、平台支持范围或具体效果数据，暂未提供。