Python 异步编程实战：用 asyncio 写出高并发 IO 应用

2026年10月22日 技术杂谈 suzhe

Python 异步编程（asyncio）是处理高并发 IO 场景的利器。当你的程序大部分时间在等待网络、数据库或文件 IO 时，asyncio 用单线程事件循环就能轻松扛住成千上万的并发连接，而不必为每个请求开一个线程。本文从协程、事件循环讲到并发原语与真实落地，带你搞清楚异步 IO 到底能做什么、不能做什么，以及怎样才能写出既快又不会踩坑的代码。

一、为什么需要异步：同步阻塞的代价

假设你有一个接口需要串行请求 3 个外部 API，每个平均耗时 300 毫秒。同步写法下，总耗时就是 300 + 300 + 300 = 900 毫秒，因为线程在等待第一个响应时什么都做不了。在 Web 服务里，这种阻塞会直接吃掉连接池和线程资源，并发一高就排队甚至超时。

asyncio 的核心思路是：当一个协程遇到 IO 等待（网络、磁盘、sleep），它主动把控制权交还事件循环，让循环去调度其他就绪的协程。于是单线程也能在等待期间”同时”推进成百上千个任务。它解决的是 IO 密集型并发，而非 CPU 密集型计算。

二、协程与 async/await 基础

在 Python 中，用 async def 定义的都是协程函数（coroutine function），调用它不会立刻执行，而是返回一个协程对象。真正的执行需要事件循环来驱动。

import asyncio

async def fetch(name, delay):
    print(f"{name} 开始，等待 {delay}s")
    await asyncio.sleep(delay)   # 模拟 IO 等待，主动让出控制权
    print(f"{name} 完成")
    return name

# asyncio.run 会创建事件循环并运行到协程结束
async def main():
    result = await fetch("任务A", 1)
    print("拿到结果：", result)

if __name__ == "__main__":
    asyncio.run(main())

关键点是 await：它只能在 async def 内部使用，表示”在这里可能要等，先把执行权交出去”。没有 await 的协程不会真正让出，也就谈不上并发。

三、并发执行：gather、wait 与任务

真正的并发来自同时调度多个协程。最常用的是 asyncio.gather，它把多个协程/任务并发跑完、按输入顺序收集结果。

async def main():
    # 三个任务"同时"开始，总耗时约等于最慢的那个（约 1s），而非 3s
    results = await asyncio.gather(
        fetch("A", 1),
        fetch("B", 1),
        fetch("C", 1),
    )
    print(results)  # ['A', 'B', 'C']

asyncio.run(main())

如果你希望”谁先完成先处理谁”，可以用 asyncio.as_completed 遍历已完成的任务；需要给单个任务设超时，则用 asyncio.wait_for。三者的取舍如下：

原语特点适用场景
gather并发收集全部结果，顺序与输入一致；任一异常会冒泡要一次性拿齐所有结果
as_completed按完成顺序逐个产出，先到先得流式处理、先完成先落库
wait_for给单个可等待对象加超时上限防止某个慢调用拖垮整体

四、异步上下文管理器与迭代器

除了函数，asyncio 还支持 async with（异步上下文管理器）和 async for（异步迭代器），这让资源管理与流式读取也变得”可等待”。

class AsyncResource:
    async def __aenter__(self):
        print("获取连接")
        return self
    async def __aexit__(self, *exc):
        print("释放连接")

async def main():
    async with AsyncResource() as r:
        await asyncio.sleep(0.1)
        print("使用资源中")

五、实战：用 aiohttp 并发抓取网页

最典型的高并发 IO 场景就是批量 HTTP 请求。下面用 aiohttp 并发抓取多个 URL，单连接也能轻松并发几十个请求。

import asyncio, aiohttp

async def get(session, url):
    async with session.get(url) as resp:
        return url, resp.status, len(await resp.text())

async def crawl(urls):
    # 限制并发，避免把目标站打挂
    connector = aiohttp.TCPConnector(limit=20)
    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [get(session, u) for u in urls]
        for coro in asyncio.as_completed(tasks):
            url, status, size = await coro
            print(f"{url} -> {status}, {size} bytes")

urls = ["https://example.com"] * 10
asyncio.run(crawl(urls))

注意 TCPConnector(limit=20) 这一步：无脑并发成百上千请求既会压垮对方，也容易触发自己的文件描述符上限。生产环境一定要做并发限流，这和我们在 SSH 连接故障排查里强调的”连接数要设上限”是同一个道理。

六、实战：异步操作数据库

关系型数据库驱动也有异步版本，例如 PostgreSQL 的 asyncpg、MySQL 的 aiomysql。它们让 Web 接口在等待查询返回时也能处理别的请求。

import asyncio, asyncpg

async def query():
    conn = await asyncpg.connect(
        user="app", password="pwd",
        database="demo", host="127.0.0.1",
    )
    rows = await conn.fetch("SELECT id, name FROM users WHERE age > $1", 18)
    await conn.close()
    return rows

async def main():
    # 并发查 3 次，总耗时接近单次而非 3 倍
    results = await asyncio.gather(query(), query(), query())
    print(sum(len(r) for r in results))

asyncio.run(main())

七、同步与异步的边界：什么时候该用、不该用

异步不是银弹。它最大的收益在 IO 密集，对 CPU 密集任务几乎无能为力——因为事件循环本身仍在单线程里跑，一段耗时的纯计算会阻塞整个循环。

任务类型是否适合 asyncio原因 / 替代方案
HTTP / 数据库 / 文件 IO非常适合大部分时间在等待，协程让出后可调度其他任务
密集计算（压缩、加密、图像处理）不适合用 multiprocessing 多进程，或 run_in_executor 卸载
调用同步阻塞库谨慎用 loop.run_in_executor 丢到线程池，避免卡住循环

当你不得不调用一个同步阻塞函数（比如某个还没支持异步的老库），可以用 run_in_executor 把它丢到默认线程池，从而不阻塞事件循环：

import asyncio, time

def block_cpu(n):
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    # 把 CPU 密集任务交给线程池，事件循环继续转
    result = await loop.run_in_executor(None, block_cpu, 10_000_000)
    print("计算结果：", result)

asyncio.run(main())

八、避坑清单

  • 别在协程里调阻塞代码：一个 time.sleep 或同步 requests 调用会卡住整个事件循环，必须换成 await asyncio.sleep 或异步库。
  • 不要为每个请求新建事件循环：在同一个程序里尽量复用 asyncio.run 或一个长期运行的 loop，循环开销虽小但新建会带来状态割裂。
  • 小心”忘记 await”：漏写 await 时代码不会报错，但协程根本不会执行，是最难排查的静默 Bug。
  • 共享状态要加锁：协程切换点可能发生在任意 await 处，跨协程共享变量请用 asyncio.Lock 而非假设”单线程就安全”。

九、和 Node.js 事件循环的关系

如果你熟悉前端或 Node.js，会发现 asyncio 的思路和 Node.js 事件循环一脉相承：都是单线程 + 事件循环 + 非阻塞 IO。区别在于 Python 用 async/await 语法把”可等待点”写得一清二楚，而 Node 早期靠回调、后来才补上 Promise 与 async 语法。理解其中一个，另一个也就触类旁通了。

十、落地清单

  • IO 密集型服务（爬虫、网关聚合、批量调用）优先评估 asyncio；
  • 并发请求务必加限流（Connector limit / Semaphore），别无脑并发；
  • 数据库、HTTP 一律用对应的异步驱动（asyncpg、aiohttp、httpx）；
  • 遇到同步阻塞库，用 run_in_executor 卸载到线程池；
  • CPU 密集任务走多进程，别指望事件循环能救场；
  • 上线前用压测量化收益，确认 P99 真的下降。

Python 异步编程的门槛不在语法，而在”什么时候用、怎么限流、如何不被阻塞调用坑到”。把上面这几条吃透，你就能用 asyncio 稳稳写出高并发的 IO 应用，而不是写出一堆更隐蔽的并发 Bug。

版权声明：
作者：suzhe
链接：https://fsdata.site/2026/10/python-asyncio-practical/
文章版权归作者所有，未经允许请勿转载。
上一篇 StampedLock 实战：高并发读多写少的乐观读锁方案
下一篇 PHP 远程代码执行漏洞：紧急自查与防护实战
最新评论
Docker 网络模式详解：bridge/host 生产选型 - FSData
Docker 网络模式详解：bridge/host 生产选型 - FSData
8月20日
[…] 注意最后那条：把端口绑定到 127.0.0.1，再由宿主机上的 Nginx 反代进来，比直接 -p 80:80 暴露安全得多，也方便统一做
评论于Nginx 反向代理完整配置：负载均衡 + HTTPS + 缓存优化
Vue3 + TypeScript 工程化：类型安全的前端架构 - FSData
Vue3 + TypeScript 工程化：类型安全的前端架构 - FSData
8月20日
[…] Pinia 的工程化用法，可以参考我们之前的实战文章 《Vue3 组合式 API + Pinia：状态管理从入门到实战》，那里覆盖了 Sto
评论于Vue3 组合式 API + Pinia：状态管理从入门到实战
2026 年最值得装的 10 个 VS Code 插件（AI/前端/后端/效率全覆盖） - FSData
2026 年最值得装的 10 个 VS Code 插件（AI/前端/后端/效率全覆盖） - FSData
8月19日
[…] 状态，点一下就能起停容器，配合我们讲过的 Docker Compose 编排 […]
评论于Docker Compose 一键编排：MySQL + Redis + 后端服务
Redis 缓存设计：穿透、击穿、雪崩与最佳实践 - FSData
Redis 缓存设计：穿透、击穿、雪崩与最佳实践 - FSData
8月19日
[…] 缓存三大坑的本质都是”缓存失效导致请求穿透到数据库”。记住口诀：穿透用布隆、击穿用锁、雪崩用错峰。配合合理的过期策略
评论于PostgreSQL 慢查询优化：执行计划解读与索引调优实战
GitHub Actions 实战：从零搭建 CI/CD 流水线 - FSData
GitHub Actions 实战：从零搭建 CI/CD 流水线 - FSData
8月19日
[…] Docker 镜像部署。结合我们之前写的 Docker Compose 一键编排实战，可以在 CI […]
评论于Docker Compose 一键编排：MySQL + Redis + 后端服务