Python 异步编程完整指南：asyncio 从入门到精通

大家好，我是程序员晚枫。

异步是 Python 的未来。

asyncio 不会用，别说会 Python。

今天这篇文章，给你 asyncio 完整指南。

一、为什么需要异步？

同步 vs 异步：

同步（普通 Python）

import time

def task1():
    time.sleep(1)
    return "Task 1"

def task2():
    time.sleep(1)
    return "Task 2"

# 串行：2 秒
start = time.time()
result1 = task1()
result2 = task2()
print(time.time() - start)  # 2 秒

异步（asyncio）

import asyncio
import time

async def task1():
    await asyncio.sleep(1)
    return "Task 1"

async def task2():
    await asyncio.sleep(1)
    return "Task 2"

# 并发：1 秒
async def main():
    start = time.time()
    results = await asyncio.gather(task1(), task2())
    print(time.time() - start)  # 1 秒

asyncio.run(main())

快 1 倍。

1000 个任务，差距更大。

二、asyncio 5 大核心概念

概念 1：协程（Coroutine）

用 async def 定义：

async def my_func():
    return 1

# 调用返回协程对象，不执行
coro = my_func()

# 执行协程
result = asyncio.run(coro)  # 1

概念 2：await

等待异步操作完成：

1
2
3

async def main():
    result = await some_async_func()
    print(result)

概念 3：任务（Task）

包装协程，让它在事件循环中运行：

async def main():
    task1 = asyncio.create_task(some_func())
    task2 = asyncio.create_task(another_func())
    
    result1 = await task1
    result2 = await task2

概念 4：Future

异步操作的"未来结果"
底层概念
一般不用直接用

概念 5：事件循环（Event Loop）

1	asyncio.run(main()) # 启动事件循环

三、5 大常用 API

API 1：asyncio.run

1	asyncio.run(main())

API 2：asyncio.gather

并发运行多个协程：

results = await asyncio.gather(
    task1(),
    task2(),
    task3()
)

API 3：asyncio.create_task

创建任务：

1 2	task = asyncio.create_task(coro) result = await task

API 4：asyncio.sleep

异步 sleep：

1	await asyncio.sleep(1) # 不阻塞

API 5：asyncio.wait_for

超时控制：

try:
    result = await asyncio.wait_for(coro, timeout=5)
except asyncio.TimeoutError:
    print("超时")

四、5 大实战场景

场景 1：异步 HTTP 客户端

import aiohttp
import asyncio

async def fetch_url(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    async with aiohttp.ClientSession() as session:
        urls = [f'https://api.example.com/{i}' for i in range(100)]
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks)

100 个请求，1 秒完成。

场景 2：异步 HTTP 服务器

from aiohttp import web

async def handle(request):
    return web.Response(text="Hello, World!")

app = web.Application()
app.router.add_get('/', handle)

web.run_app(app)

20 行 = 异步 HTTP 服务器。

场景 3：异步数据库

import asyncpg
import asyncio

async def main():
    conn = await asyncpg.connect('postgresql://...')
    rows = await conn.fetch('SELECT * FROM users')
    for row in rows:
        print(row)
    await conn.close()

场景 4：异步 WebSocket

import asyncio
import websockets

async def echo(websocket):
    async for message in websocket:
        await websocket.send(f"Echo: {message}")

async def main():
    async with websockets.serve(echo, "localhost", 8765):
        await asyncio.Future()  # 永久运行

asyncio.run(main())

场景 5：异步队列

import asyncio

async def producer(queue):
    for i in range(10):
        await queue.put(i)
        await asyncio.sleep(0.1)

async def consumer(queue):
    while True:
        item = await queue.get()
        print(f"Consumed: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue()
    await asyncio.gather(
        producer(queue),
        consumer(queue)
    )

五、5 大常见陷阱

陷阱 1：在协程中用 time.sleep

# 错：阻塞事件循环
async def bad():
    time.sleep(1)  # 阻塞

# 对：异步 sleep
async def good():
    await asyncio.sleep(1)  # 不阻塞

陷阱 2：忘记 await

# 错：返回协程对象，不执行
result = my_coro()

# 对：必须 await
result = await my_coro()

陷阱 3：在同步代码中调异步

# 错：在同步函数中调异步
def sync_func():
    result = asyncio.run(my_coro())  # 可以但不优雅

# 对：顶层入口用 asyncio.run
async def main():
    result = await my_coro()

asyncio.run(main())

陷阱 4：CPU 密集任务用 asyncio

asyncio 适合 I/O 密集
CPU 密集用 multiprocessing

陷阱 5：混用同步和异步库

同步库阻塞事件循环
用异步版本（如 httpx 替代 requests）

六、5 大性能优化技巧

技巧 1：使用 uvloop

替代默认事件循环：

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

asyncio.run(main())

快 2-4 倍。

技巧 2：使用连接池

import aiohttp

connector = aiohttp.TCPConnector(limit=100)
async with aiohttp.ClientSession(connector=connector) as session:
    # 复用连接
    pass

技巧 3：使用 Semaphore 限流

sem = asyncio.Semaphore(10)

async def fetch(url):
    async with sem:
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                return await response.text()

技巧 4：避免阻塞操作

# 错
import time
async def bad():
    time.sleep(1)

# 对
async def good():
    await asyncio.sleep(1)

技巧 5：使用异步文件 I/O

import aiofiles

async with aiofiles.open('file.txt', 'r') as f:
    content = await f.read()

七、5 大异步框架

框架 1：FastAPI

最快 Python Web 框架
原生异步
自动文档

框架 2：aiohttp

HTTP 客户端/服务器
异步 Web 框架

框架 3：Tornado

老牌异步
实时 Web

框架 4：Starlette

FastAPI 底层
轻量异步

框架 5：Sanic

类 Flask
异步

八、5 个真实项目案例

案例 1：OpenAI API

使用：aiohttp + asyncio
结果：服务百万开发者

案例 2：Dropbox 同步

使用：asyncio + aiohttp
结果：5 亿用户

案例 3：Instagram 实时

使用：asyncio + websockets
结果：10 亿用户

案例 4：Slack 消息

使用：asyncio + websockets
结果：千万级企业

案例 5：Discord

使用：asyncio 全栈
结果：千万级用户

九、5 个常见误区

误区 1：asyncio 一定快

❌ 错
✅ 只对 I/O 密集 有效
CPU 密集用 multiprocessing

误区 2：asyncio 很难

⚠️ 部分对
✅ 概念多
1 周上手

误区 3：所有项目都该用 asyncio

❌ 错
✅ 简单脚本不需要
复杂服务才需要

误区 4：asyncio 替代多线程

⚠️ 部分对
✅ asyncio 是单线程
多核用多进程

误区 5：asyncio 替代多进程

❌ 错
✅ asyncio 单线程
CPU 密集必须多进程

十、5 个学习路径

路径 1：完全新手

1	async/await 基础（1 周）→ asyncio.gather（1 周）→ 简单项目

路径 2：Web 开发者

1	FastAPI（1 周）→ 异步数据库（1 周）→ 实战项目

路径 3：爬虫开发者

1	aiohttp 客户端（1 周）→ 异步爬虫项目（2 周）

路径 4：网络开发者

1	asyncio TCP/UDP（1 周）→ websockets（1 周）→ 实战项目

路径 5：AI 工程师

1	FastAPI（1 周）→ 异步推理（2 周）→ 模型部署

十一、给 Python 异步学习者的 4 个建议

建议 1：先学 async/await

5 天上手
理解协程

建议 2：用真实项目练

异步爬虫
异步 API
1 个月

建议 3：理解 I/O 密集

知道什么时候用
关键

建议 4：不要乱用

简单脚本不需要
按需用

十二、最后的最后

Python asyncio，3 句话总结：

asyncio 是 Python 的未来：异步是趋势
async/await 是核心：简单
I/O 密集用 asyncio：CPU 密集用多进程

学 Python 6 年，我学到的最重要的事：

"不会 asyncio，别说会 Python。"

5 年后，asyncio 会和列表推导式一样普及。

现在学 asyncio，5 年后你就是专家。

🎓 AI 编程实战课程

程序员晚枫专注AI编程培训，通过 《50讲 · AI编程训练营》，让小白也能用AI做出实际项目。帮你从零上手！

👉 免费试看：网盘链接，免费试看前3讲，先看看适不适合自己
👉 - 👉 课程报名：点击这里报名，现在报名还送书📖