PorterYuan · 个人学习博客

常见坑和实战模式

8分钟阅读

常见坑和实战模式

一、常见坑和实战模式

1.1 坑 ①：async 函数里调同步阻塞代码

最常犯的错。

 
# 在 async 函数里调用 time.sleep()——整个事件循环被堵死！
 
async def bad_handler():
 
    time.sleep(3)              # 同步阻塞！事件循环在等 sleep 结束——什么都干不了
 
    return "done"
 
# ✅ 用 asyncio.sleep()——把控制权交还给事件循环
 
async def good_handler():
 
    await asyncio.sleep(3)     # 挂起 3 秒——事件循环可以去执行其他协程
 
    return "done"
 
 
time.sleep(3)：                  # 线程睡了 3 秒——真睡
 
asyncio.sleep(3) + await：       # 告诉事件循环"3 秒后叫我"——自己暂停，别人继续

原则：async 函数里永远不要直接调同步阻塞函数。要么用异步版（aiohttp 代替 requests、aiomysql 代替 pymysql），要么丢到线程池。

1.2 坑 ②：创建协程但不 await

 
# ❌ 只创建不 await——协程不会执行！
 
async def main():
 
    download("url1")             # ⚠️ 创建了协程对象，但没 await → 警告 + 不会执行
 
    download("url2")             # 同上
 
    # RuntimeWarning: coroutine 'download' was never awaited
 
# ✅ 方案一：直接 await（串行）
 
async def main():
 
    await download("url1")
 
    await download("url2")
 
# ✅ 方案二：create_task（并发）
 
async def main():
 
    task1 = asyncio.create_task(download("url1"))
 
    task2 = asyncio.create_task(download("url2"))
 
    result1 = await task1
 
    result2 = await task2

1.3 坑 ③：忘了 await Task

 
# ❌ create_task 后忘了 await——主协程结束，子 Task 被取消
 
async def main():
 
    task = asyncio.create_task(download("slow_url"))
 
    return "main done"            # main 结束了！task 可能还没跑完——被取消
 
    # ⚠️ task 被 GC 回收或事件循环关闭时丢掉
 
# ✅ 确保 await 所有 Task
 
async def main():
 
    task = asyncio.create_task(download("slow_url"))
 
    result = await task           # 等到 task 完成再返回
 
    return "main done"
 
# ⚠️ 特殊场景：Task 可以"放飞"——但不推荐
 
async def main():
 
    task = asyncio.create_task(background_work())  # 后台任务
 
    # 不 await——task 会继续跑
 
    # 风险：如果 main 所在的 asyncio.run() 结束，task 也被取消

1.4 坑 ④：在已有事件循环的环境里调 asyncio.run

 
# ❌ 在 Jupyter / FastAPI 路由里不能用 asyncio.run()
 
async def endpoint():
 
    asyncio.run(some_coro())     # RuntimeError: asyncio.run() cannot be called from a running event loop
 
# ✅ 直接用 await
 
async def endpoint():
 
    return await some_coro()     # 已在事件循环中——直接 await 就行

1.5 实战模式总结

 
# 模式 1：最简——一个 async 函数搞定
 
async def main():
 
    data = await fetch()
 
    result = await process(data)
 
    return result
 
asyncio.run(main())
 
# 模式 2：并发 N 个相同任务
 
async def main():
 
    tasks = [fetch(url) for url in urls]
 
    results = await asyncio.gather(*tasks)
 
# 模式 3：一个生产者 + N 个消费者
 
async def producer(queue):
 
    for item in items:
 
        await queue.put(item)
 
    await queue.put(None)       # 毒丸——告诉消费者"没了"
 
async def consumer(queue):
 
    while True:
 
        item = await queue.get()
 
        if item is None:
 
            break
 
        await process(item)
 
async def main():
 
    queue = asyncio.Queue(maxsize=10)        # 异步队列——await put/get
 
    await asyncio.gather(
 
        producer(queue),
 
        consumer(queue),
 
        consumer(queue),                    # 两个消费者并行
 
    )
 
# 模式 4：超时 + 重试
 
async def fetch_with_retry(url, retries=3, timeout=5):
 
    for i in range(retries):
 
        try:
 
            return await asyncio.wait_for(download(url), timeout=timeout)
 
        except asyncio.TimeoutError:
 
            if i == retries - 1:
 
                raise
 
            await asyncio.sleep(2 ** i)    # 指数退避：1s, 2s, 4s
 
# 模式 5：优雅关闭——收到信号后等现有任务完成
 
async def main():
 
    tasks = [asyncio.create_task(work(i)) for i in range(10)]
 
    try:
 
        await asyncio.gather(*tasks)
 
    except asyncio.CancelledError:
 
        for t in tasks:
 
            t.cancel()                                   # ① 标记"请取消"
 
        await asyncio.gather(*tasks, return_exceptions=True)  # ② 等所有 task 做完清理
 
    # cancel 不是立刻停——等 task 执行到下一个 await 时才抛 CancelledError
 
    # 目的：给 task 机会清理资源（关闭连接、保存状态）

1.6 坑 ⑤：CPU 密集型后台任务堵事件循环

create_task 不开新线程——还是在事件循环线程里跑。

 
# ❌ CPU 密集型协程——没有 await，一直占着事件循环线程
 
async def heavy_compute():
 
    for i in range(10_000_000):
 
        x = i * i       # 没有 await——不交出控制权
 
    return x
 
async def main():
 
    task = asyncio.create_task(heavy_compute())  # 还是事件循环线程跑！
 
    await fetch()                                 # ❌ 等不到——heavy_compute 不放手
 
# ✅ 丢进程池——另一个进程跑，不占事件循环
 
async def main():
 
    loop = asyncio.get_running_loop()
 
    result = await loop.run_in_executor(
 
        ProcessPoolExecutor(), heavy_compute     # 进程池，不是线程池
 
    )

一句话：create_task 只是注册到事件循环——不开新线程。CPU 密集型必须 run_in_executor 丢进程池。

1.7 坑 ⑥：cancel 不是立刻停

 
task.cancel()   # 只是设标志位——不立刻杀
 
 
cancel 之后：
 
  协程正在同步执行（没 await）→ 等执行到下一个 await 时才抛 CancelledError
 
  协程正在 await              → 立刻抛 CancelledError
 
  协程暂停中（等 I/O）         → 唤醒，抛 CancelledError
 
为什么不能立刻杀？
 
  立刻杀 = 资源泄漏（连接没关、文件没关）
 
  等 await = 给协程机会 catch 异常 → 自己清理 → 安全退出

速记卡（面试闪卡）

Q1：一句话讲清「常见坑和实战模式」到底是什么？

A：协程两大坑：async函数里调同步阻塞代码（如time.sleep）会堵死整个事件循环；只创建协程不await则根本不执行——要用asyncio.sleep、create_task或丢线程池。

Q2：坑一：同步阻塞堵死事件循环 —— 怎么理解？

A：async 函数里调 time.sleep() 像在单行道上停车睡觉：事件循环（Event Loop）这一个线程被你占住，所有其他协程全卡死。必须用 asyncio.sleep() 把控制权交还循环，别人才能跑。

Q3：坑二：创建协程不 await —— 怎么理解？

A：只写 download(“url”) 不 await，像买了电影票却不去看——协程对象创建了但永不执行，还会抛 RuntimeWarning。必须 await 或用 asyncio.create_task 把它交给循环调度。

Q4：解法：把阻塞丢进线程池 —— 怎么理解？

A：遇到非得用的同步库，用 asyncio.to_thread() 或 loop.run_in_executor 把阻塞调用丢进线程池（ThreadPool），事件循环腾出来继续跑协程，阻塞在别的线程慢慢睡。

Q5：create_task：并发的开关 —— 怎么理解？

A：asyncio.create_task() 把协程”登记”进事件循环并发跑，像同时按下多个开关；配合 gather 等它们全完成。和单纯 await（串行）不同，create_task 才是真正并发的关键。

Q6：核心速记主线有哪些？

  • 坑一：async 内调同步阻塞会堵死事件循环

  • 坑二：只建协程不 await 永不执行

  • 解法：asyncio.sleep 让权、to_thread 跑阻塞

  • create_task + gather 实现真正并发

口诀

A：同步阻塞堵一路

建了不await白搭

让权线程丢阻塞

任务登记才并发

相关链接

关系图谱

反向链接

近期更新