Ubuntu Python异步编程入门

Ubuntu 下 Python 异步编程入门指南

一 环境准备

想在 Ubuntu 上玩转异步编程，首先得把环境搭好。这里有几个关键步骤：

• Python 版本选择：强烈建议使用 Python 3.7 或更高版本。如果你的系统版本比较旧，可以通过添加 deadsnakes PPA 源来安装新版本。别忘了创建虚拟环境，它能帮你把项目依赖隔离得干干净净。
  • 安装与准备
    • 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 语法则在 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())

• 运行方式
  • 记住这个标准流程：用 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，把阻塞函数丢到线程池或进程池里去执行。