0Pricing
Python Academy · 课时

生产环境中的异步 I/O 模式

将 asyncio 应用于 HTTP 客户端、队列和实际生产环境中的数据处理流程。

生产环境中的异步 I/O 模式 是 CoddyKit 上的免费 Python Academy 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 Python Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 Python Academy 课程共包含 4 节课。

事件循环最佳实践

始终使用 asyncio.run() 作为唯一入口点。避免不谨慎地混用同步和异步代码，切勿让长时间运行的 CPU 工作阻塞事件循环。

import asyncio

async def main():
    # Never: time.sleep() — blocks the event loop
    # Always: await asyncio.sleep()
    await asyncio.sleep(1)

asyncio.run(main())

连接池

请重复使用连接，而不是为每个请求打开新连接。asyncpg 和 aiohttp 等库提供了内置连接池。

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@host/db",
        min_size=5, max_size=20
    )
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT 1")
    await pool.close()

使用 asyncio.Queue 实现生产者—消费者模式

asyncio.Queue 将作为并发任务运行的生产者和消费者解耦。

import asyncio

async def producer(q):
    for i in range(5):
        await q.put(i)
        await asyncio.sleep(0.5)
    await q.put(None)  # sentinel

async def consumer(q):
    while (item := await q.get()) is not None:
        print(f"processing {item}")
        q.task_done()

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

asyncio.run(main())

使用 Semaphore 进行速率限制

限制并发的出站请求数量，以免使外部 API 不堪重负或触发速率限制。

import asyncio, aiohttp

SEM = asyncio.Semaphore(10)  # 10 concurrent requests max

async def fetch(session, url):
    async with SEM:
        async with session.get(url) as r:
            return await r.json()

async def main(urls):
    async with aiohttp.ClientSession() as s:
        return await asyncio.gather(*[fetch(s, u) for u in urls])

带指数退避的重试

为可重试的操作封装退避逻辑，以便平稳地处理暂时性故障。

import asyncio

async def retry(coro_fn, retries=3, base=1):
    for attempt in range(retries):
        try:
            return await coro_fn()
        except Exception as e:
            if attempt == retries - 1: raise
            delay = base * 2 ** attempt
            await asyncio.sleep(delay)

优雅关闭

在长时间运行的任务中捕获 asyncio.CancelledError，并处理 OS 信号，以实现干净的关闭。

import asyncio, signal

async def server():
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, stop.set)
    await stop.wait()
    print("Shutting down gracefully")

流式响应

使用异步生成器流式传输大型响应，而不必将其完整加载到内存中。

import asyncio, aiohttp

async def stream(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            async for chunk in r.content.iter_chunked(4096):
                process(chunk)

async def process(chunk):
    await asyncio.sleep(0)
    print(len(chunk), "bytes")

FastAPI 中的后台任务

FastAPI 的 BackgroundTasks 会将工作排队，以便在响应发送后运行，从而让端点保持快速响应。

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, body: str):
    # heavy synchronous work
    pass

@app.post("/register")
async def register(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email, "Welcome!")
    return {"status": "registered"}

心跳模式

使用 create_task，在主服务器循环旁运行周期性的后台任务。

import asyncio

async def heartbeat():
    while True:
        print("alive")
        await asyncio.sleep(30)

async def main():
    asyncio.create_task(heartbeat())
    await serve_forever()

分析异步代码

使用 yappi 或 py-spy 进行支持异步代码的性能分析。启用 asyncio 调试模式，以获取慢回调警告。

# Enable debug to catch slow callbacks:
# asyncio.run(main(), debug=True)

# PYTHONASYNCIODEBUG=1 python server.py
# logs: Executing <Task coroutine=...> took 0.150 seconds

避免常见陷阱

关键规则：切勿在事件循环中调用阻塞代码；始终等待协程；关闭时取消任务；使用连接池；使用信号量限制并发。

import asyncio

async def bad():
    import time
    time.sleep(2)        # blocks the entire event loop!

async def good():
    await asyncio.sleep(2)  # yields control

快速检查

哪个 asyncio 原语可以将作为并发任务运行的生产者和消费者解耦？

回顾

生产环境中的异步模式：使用连接池，使用 Semaphore 控制并发，实现重试和退避，流式处理大型数据，在信号到达时优雅关闭，并使用 create_task 运行周期性任务。

常见问题解答

「生产环境中的异步 I/O 模式」课时是免费的吗？

是的 — 「生产环境中的异步 I/O 模式」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 Python Academy 课程的其余内容，请升级到 CoddyKit PRO。 Python Academy 课程共包含 4 节课。

「生产环境中的异步 I/O 模式」这节课中我会学到什么？

将 asyncio 应用于 HTTP 客户端、队列和实际生产环境中的数据处理流程。 你通过在浏览器中直接运行的动手代码来练习 Python Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 Python Academy 需要有经验吗？

无需任何先前经验。CoddyKit 上的 Python Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。

「生产环境中的异步 I/O 模式」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 Python Academy 课中编写并运行代码吗？

能。每节 Python Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 协程与事件循环
  2. await、任务与汇集
  3. 异步上下文管理器与迭代器
  4. 生产环境中的异步 I/O 模式
← 返回 Python Academy