Ubuntu 下 Python 异步编程入门指南
一 环境准备
想在 Ubuntu 上玩转异步编程，首先得把环境搭好。这里有几个关键步骤：
- Python 版本选择：强烈建议使用 Python 3.7 或更高版本。如果你的系统版本比较旧，可以通过添加 deadsnakes PPA 源来安装新版本。别忘了创建虚拟环境，它能帮你把项目依赖隔离得干干净净。
- 安装与准备
sudo apt update && sudo apt install -y python3 python3-pipsudo add-apt-repository ppa:deadsnakes/ppasudo apt install -y python3.10 python3.10-venv
- 创建虚拟环境
python3.10 -m venv .venvsource .venv/bin/activate
- 安装与准备
- 常用异步库
pip install aiohttp httpx
简单说明一下：asyncio 模块从 Python 3.4 开始引入，而我们现在熟悉的 async/await 语法则在 3.5 版本稳定下来。之所以推荐 3.7+，是因为从这个版本开始，提供了像 asyncio.run() 这样更简洁、更友好的程序入口，对新手来说门槛低了不少。
二 核心概念与第一个异步程序
理解了下面几个概念，异步编程的大门就算推开一半了。
- 关键概念
- 事件循环 (Event Loop)：它是整个异步运行的心脏，负责调度和切换各个协程。
- 协程 (Coroutine)：用
async def定义的函数，它的妙处在于能在等待时主动“让出”控制权。 - 任务 (Task)：可以理解为对协程的包装，交给事件循环去并发调度。
- 可等待对象 (Awaitable)：这是一个统称，协程、任务和 Future 都属于此类。
- 最小示例
- 顺序执行：让两个协程A和B依次运行，分别等待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())
-
- 顺序执行：让两个协程A和B依次运行，分别等待1秒和2秒，总耗时自然是3秒左右。
- 运行方式
- 记住这个标准流程：用
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(r) except asyncio.TimeoutError: print(“timeout”) asyncio.run(main())
-
- 任务生命周期：用
create_task创建任务后，务必记得通过await或者在gather中等待它完成。否则，主协程一旦提前退出，那些还没跑完的任务就会被无情取消。 - 何时使用异步 vs 线程/进程
- I/O 密集型（比如网络请求、磁盘读写、数据库操作）：这是异步编程的主场，优先选择 asyncio。
- CPU 密集型（比如大量数学计算）：异步在这里帮不上忙，反而可能因为事件循环被阻塞而变慢。这种情况下，应该优先考虑
multiprocessing或者concurrent.futures.ProcessPoolExecutor。当然，也可以在事件循环里用run_in_executor，把阻塞函数丢到线程池或进程池里去执行。
