SuperAPI AI 工具集
合作ENAI 行业资讯收藏
Skill 广场asyncio

asyncio

编程与开发

查看项目

更新于 10-15 12:18

技能介绍

用途

Python asyncio 是 Python 标准库中用于编写并发代码的模块，采用 async/await 语法。它适合 I/O 密集型操作，例如 HTTP 请求、数据库查询、文件操作和 WebSocket 连接。asyncio 提供非阻塞执行能力，无需引入线程或多进程的复杂度。

主要能力

  • async/await 语法，用于编写可读的并发代码
  • 事件循环，用于管理并发操作
  • 任务（Task），用于并发运行多个协程
  • 同步原语：锁、信号量、事件、队列、条件变量
  • 基于 aiohttp 的 HTTP 客户端与服务端
  • 数据库异步支持（asyncpg、aiomysql、motor）
  • FastAPI 异步端点
  • WebSocket 支持
  • 后台任务管理

安装

asyncio 为内置模块（Python 3.7+），无需安装。配套库可按需安装：

# 异步 HTTP 客户端
pip install aiohttp

# 异步 HTTP 请求（替代方案）
pip install httpx

# 异步数据库驱动
pip install asyncpg aiomysql motor  # PostgreSQL, MySQL, MongoDB

# 支持异步的 FastAPI
pip install fastapi uvicorn[standard]

# 异步测试
pip install pytest-asyncio

使用方式

基础 async/await 模式

  1. 简单异步函数：使用 async def 定义协程函数，await 挂起执行直到可等待对象完成，asyncio.run() 作为异步程序入口。协程必须被 await 或调度。

  2. 多任务并发：使用 asyncio.gather() 并发运行多个任务。示例中三个任务分别耗时 3、2、1 秒，顺序执行需 6 秒，并发执行仅需 3 秒。

  3. 任务创建与管理：使用 asyncio.create_task() 创建任务（立即开始），在任务运行期间可执行其他工作，最后 await 任务获取结果。

  4. 异步错误处理：可使用 try/except 捕获单个协程异常；asyncio.gather() 配合 return_exceptions=True 可返回异常而不中断其他任务。

事件循环基础

  1. 事件循环生命周期：现代方式使用 asyncio.run(main()) 自动创建循环、运行主协程并关闭循环；高级场景可手动管理循环，使用 loop.create_task() 调度协程、loop.call_later() 调度回调。asyncio.get_running_loop() 获取当前事件循环。

  2. 循环调度与回调：loop.call_soon() 调度立即回调，loop.call_later() 延迟回调，loop.call_at() 在指定时间回调。

  3. 运行阻塞代码：使用 loop.run_in_executor(None, blocking_io) 在线程池中运行阻塞代码，避免阻塞事件循环；可配合 asyncio.gather() 并发运行多个阻塞操作。

Asyncio 同步原语

  1. 锁（Lock）：用于互斥，保护共享资源。示例对比了无锁时的竞态条件（计数结果常小于 3）与有锁时的正确结果（始终为 3）。

  2. 信号量（Semaphore）：限制并发操作数量。asyncio.Semaphore(2) 表示最多 2 个并发，创建 5 个任务时仅 2 个同时持有信号量。

  3. 事件（Event）：用于信号通知。event.wait() 阻塞直到事件被设置，event.set() 唤醒所有等待者。

  4. 队列（Queue）：用于任务分发。生产者通过 queue.put() 添加项目，消费者通过 queue.get() 获取，queue.task_done() 标记完成，queue.join() 等待所有项目处理完毕。

  5. 条件变量（Condition）：condition.wait_for() 等待条件满足，condition.notify(n=1) 通知一个等待者，condition.notify_all() 通知全部。

使用 aiohttp 进行异步 HTTP

  1. 基础 HTTP 客户端：使用 aiohttp.ClientSession() 创建会话，session.get(url) 发起请求，response.text() 读取文本，配合 asyncio.gather() 并发请求多个 URL。

  2. 带错误处理的 HTTP 客户端：实现重试逻辑，使用 response.raise_for_status() 对 4xx/5xx 抛出异常，捕获 aiohttp.ClientError 和 asyncio.TimeoutError，采用指数退避（await asyncio.sleep(2 ** attempt)）。

  3. HTTP 服务端：使用 aiohttp.web 创建应用，app.router.add_get() 和 add_post() 注册路由，支持 JSON 响应、POST JSON 请求体处理、流式响应（web.StreamResponse），通过 web.run_app(app, host="0.0.0.0", port=8080) 运行。

  4. WebSocket 客户端：使用 session.ws_connect(url) 连接，ws.send_str() 和 ws.send_json() 发送消息，通过 async for msg in ws 接收消息，处理 aiohttp.WSMsgType.TEXT 等消息类型。

适用场景

  • I/O 密集型任务：HTTP 请求、数据库查询、文件操作、WebSocket 连接
  • 需要高并发但不想使用线程或多进程的场景
  • 异步 Web 服务（aiohttp、FastAPI）
  • 异步数据库访问（PostgreSQL、MySQL、MongoDB）
  • 需要限制并发数量或协调多个协程的场景

限制

  • 资料未提供关于 CPU 密集型任务适用性的说明
  • 资料未提供性能基准或与其他并发方案的对比数据
  • 资料未提供版本兼容性细节（除 Python 3.7+ 外）
  • 资料未提供生产环境部署或调优建议
  • 资料未提供收费、价格或平台相关信息