Ubuntu Python异步编程入门

Ubuntu 下 Python 异步编程入门指南

一 环境准备

• 在 Ubuntu 上建议使用 Python 3.7+。如系统版本较旧，可添加 deadsnakes PPA 安装新版 Python，并创建虚拟环境隔离依赖：
  • 安装与准备
    • sudo apt update && sudo apt install -y python3 python3-pip
    • sudo add-apt-repository ppa:deadsnakes/ppa
    • sudo apt install -y python3.10 python3.10-venv
  • 创建虚拟环境
    • python3.10 -m venv .venv
    • source .venv/bin/activate
• 常用异步库
  • pip install aiohttp httpx
• 说明：asyncio 自 Python 3.4 引入，async/await 语法自 Python 3.5 稳定可用；本入门以 3.7+ 为主，因其提供 asyncio.run 等更友好的入口。

二 核心概念与第一个异步程序

• 关键概念
  • 事件循环 Event Loop：负责调度与切换协程，是异步运行的核心。
  • 协程 Coroutine：用 async def 定义，可在等待时让出控制权。
  • 任务 Task：对协程的包装，交由事件循环并发调度。
  • 可等待对象 Awaitable：包括协程、任务、Future。
• 最小示例
  • 顺序执行：两个协程各等待 1 秒 与 2 秒，总耗时约 3 秒。
    • import asyncio async def say_delay(name, sec): print(f"[{name}] start") await asyncio.sleep(sec) print(f"[{name}] done") async def main(): await say_delay(“A”, 1) await say_delay(“B”, 2) if name == “main”: asyncio.run(main())
  • 并发执行：使用 asyncio.create_task 并发两个协程，总耗时约 2 秒（取最大等待）。
    • import asyncio async def say_delay(name, sec): print(f"[{name}] start") await asyncio.sleep(sec) print(f"[{name}] done") async def main(): t1 = asyncio.create_task(say_delay(“A”, 1)) t2 = asyncio.create_task(say_delay(“B”, 2)) await t1; await t2 if name == “main”: asyncio.run(main())
• 运行方式
  • 使用 asyncio.run(main()) 作为程序入口；在协程内部使用 await 等待可等待对象；通过 create_task 将协程调度为任务以获得并发。

三 实战 异步 HTTP 请求

• 使用 aiohttp 并发抓取多个站点
  • import asyncio, aiohttp async def download_site(url, session): async with session.get(url) as resp: print(f"{url} -> {resp.status}") async def main(): urls = [“https://example.com”, “https://example.org”] async with aiohttp.ClientSession() as session: tasks = [download_site(u, session) for u in urls] await asyncio.gather(*tasks) if name == “main”: asyncio.run(main())
• 使用 httpx（支持同步/异步）
  • pip install httpx
  • import asyncio, httpx async def fetch(url): async with httpx.AsyncClient() as client: r = await client.get(url) return r.status_code async def main(): urls = [“https://www.example.com”, “https://www.python.org”, “https://www.github.com”] results = await asyncio.gather(*[fetch(u) for u in urls]) for u, s in zip(urls, results): print(u, s) if name == “main”: asyncio.run(main())

四 常见陷阱与最佳实践

• 避免阻塞调用：在协程中不要使用 time.sleep、requests.get 等阻塞 API；改用 await asyncio.sleep() 与异步 HTTP 客户端（如 aiohttp/httpx）。
• 控制并发度：大量并发可能压垮目标服务或本机资源，使用 asyncio.Semaphore 限流。
  • import asyncio async def worker(i, sem): async with sem: print(f"worker-{i} working") await asyncio.sleep(1) async def main(): sem = asyncio.Semaphore(5) # 最多同时 5 个 await asyncio.gather(*[worker(i, sem) for i in range(20)]) asyncio.run(main())
• 超时与容错：对可能慢或不可靠的操作使用 asyncio.wait_for 设置超时，并妥善捕获异常。
  • import asyncio async def job(): await asyncio.sleep(5) return “ok” async def main(): try: r = await asyncio.wait_for(job(), timeout=2) print® except asyncio.TimeoutError: print(“timeout”) asyncio.run(main())
• 任务生命周期：使用 create_task 后，务必 await 或在 gather 中等待，避免主协程提前退出导致任务被取消。
• 何时使用异步 vs 线程/进程
  • I/O 密集型（网络、磁盘、数据库）：优先 asyncio。
  • CPU 密集型：优先 multiprocessing 或 concurrent.futures.ProcessPoolExecutor；也可在事件循环中用 run_in_executor 将阻塞函数丢到线程/进程池执行。