Ubuntu Python异步编程入门

数码系统

相机 win10

测评 win11

手机智车

华为 Tesla

小米理想

苹果蔚来

游戏软件

LOL 抖音

原神微信

当前位置：首页

编程语言

Ubuntu Python异步编程入门

热心网友时间：2026-05-05

转载

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，把阻塞函数丢到线程池或进程池里去执行。