异步函数与异步生成器这对概念我琢磨了挺长时间才真正吃透。说实话网上的教程一抓一大把但绝大多数都停留在“讲语法”的层面看完你会用async def和await但一碰到真实场景比如高并发请求、流式数据处理还是不知道怎么组织代码。这篇文章我不打算给你念文档咱就从一个实际写异步程序的人的角度聊聊这两个东西到底是什么、怎么用、踩过哪些坑。这篇内容适合这几类人看刚学完 Python 基础、想搞懂 asyncio 到底怎么回事的初学者已经会用 requests 但被性能瓶颈卡住、想切 aiohttp 的爬虫开发者以及写后端服务时被阻塞调用拖垮、想用异步优化但不知道怎么设计代码的朋友。核心就三件事异步函数怎么正确声明和调用、异步生成器怎么帮你处理流式数据、以及这两者在真实项目里怎么配合。1. 先搞清楚什么是异步函数不是“快”而是“不等”1.1 为什么同步代码会卡住要理解异步函数得先理解同步代码的痛点。你写一个普通的函数它会从头到尾顺序执行碰到耗时的操作网络请求、文件读写、数据库查询就傻乎乎地等结果这个等待的时间里CPU 什么事也没干。更糟糕的是如果你在 Web 服务里这么写一个请求卡在 I/O 上整个进程的其它请求都得陪着等。我最早写爬虫的时候就是这种体验用 requests 循环发请求100 个 URL 要一个个等响应少说几十秒。那会儿我以为是网络慢后来才明白问题根本不在网速是我的代码在“傻等”。1.2 异步函数的本质协作式调度异步函数的核心机制是协作式调度四个字拆开解释协作每个任务自觉地在 I/O 等待点“让出”控制权而不是被操作系统强行打断那是抢占式调度。调度一个叫事件循环Event Loop的东西负责管理所有任务谁可以继续执行、谁还得等都是它说了算。用生活化的类比来说同步编程像你在食堂窗口排队打饭你得等前面的人打完才轮到你异步编程像你点完单拿了个号先去别的窗口吃小吃等餐好了再回来取。你并没有“同时”做很多事但你的时间没有浪费在干等上。注意异步不是多线程它是单线程内的并发。异步函数看起来是“同时”跑的底层其实是事件循环在快速切换同一时刻只有一段代码在执行。1.3 async def 和 await 的“搭配规矩”定义异步函数很简单把def换成async def就够了。但真正的小坑在于await的使用规则我见过无数新手在这里翻车。async def fetch_data(): resp await http_get(https://example.com/api) data resp.json() return data三条红线必须记住await只能在async def函数内部使用在普通函数里用await直接语法错误。await的对象必须是“可等待对象”包括协程coroutine、任务Task、Future 这三种。调用一个异步函数不会执行函数体只是创建一个协程对象。你必须有await或把它交给事件循环函数体才会真正跑起来。这最后一条是新手最容易懵的地方。很多人这么写async def main(): fetch_data() # 错这里只是创建了协程对象函数根本没执行正确的写法是await fetch_data()或者asyncio.create_task(fetch_data())。这个区别我后面还会反复提到因为它是理解整个异步编程模型的钥匙。2. 从函数到任务让多个异步函数真正跑起来2.1 await 是“等”不是“并行”很多初学者误以为写了async def就自动并行了其实大错特错。看这段代码import asyncio async def task_a(): print(A start) await asyncio.sleep(2) print(A end) async def task_b(): print(B start) await asyncio.sleep(1) print(B end) async def main(): await task_a() await task_b() asyncio.run(main())这个程序总耗时是 3 秒不是 2 秒。因为main里先await task_a()意思是“等 task_a 执行完再说”task_a 结束前 task_b 根本没机会跑。注意await task_a()时控制权确实回到了事件循环但main自己也在等待事件循环里除了 task_a 没有别的任务可切换所以只能干等。sleep(2)结束之前 task_b 完全没有被调度。这就是传说中的“有 await 但不并发”。2.2 create_task 和 gather并发从哪来真正让多个异步函数并发跑的办法是让它们成为“任务”Task或者用gather、asyncio.wait这类高层 API 把它们分组。async def main(): # 方式一手动创建任务 a_task asyncio.create_task(task_a()) b_task asyncio.create_task(task_b()) await a_task await b_task # 方式二gather 一行搞定 await asyncio.gather(task_a(), task_b())create_task的作用是把协程包装成 Task并立刻把它注册到事件循环的调度队列里。这样main进行到await a_task时虽然它自己不往下走但事件循环里已经有 task_a 和 task_b 两个任务了task_a 每次await asyncio.sleep()让出控制权时事件循环就会切到 task_b 去跑。两个任务的耗时就被重叠起来了。用gather更省事它内部帮你处理了任务创建和结果收集返回值是对应协程结果的列表。results await asyncio.gather( fetch_data(url1), fetch_data(url2), fetch_data(url3), ) # results [data1, data2, data3]关键认知并发是任务级别的不是函数定义级别的。同样是两个异步函数写await f1()再await f2()就是串行用create_task包一层就是并发。差别就藏在你怎么调用它们。2.3 超时与取消异步里的“刹车型号”并发跑起来了新问题跟着就来了如果某个任务卡住了怎么办同步代码里你最多就是等但异步场景下一个任务卡死可能拖垮整个事件循环。asyncio.wait_for给 await 加了时限try: result await asyncio.wait_for(fetch_data(url), timeout3) except asyncio.TimeoutError: # 处理超时 result fallback_data()wait_for内部会在超时后取消对应的任务所以它不仅是“等不到就放弃”还会主动给任务发取消信号。如果你的异步函数里写了finally块就有机会做资源清理。取消机制同样可以直接用task asyncio.create_task(fetch_data(url)) await asyncio.sleep(1) task.cancel() try: await task except asyncio.CancelledError: pass这里要特别提醒asyncio.CancelledError在 Python 3.8 之后不再继承自Exception而是继承自BaseException。所以如果你的代码里有except Exception是接不住取消信号的。这算是一个隐蔽的坑。3. 异步生成器把“懒加载”和“异步”结合起来3.1 为什么要用异步生成器先说普通生成器。yield让函数变成迭代器每次next()才执行到下一个yield这种“惰性求值”在数据处理时能省内存。异步生成器就是把yield和await结合起来。函数里既能await异步操作又能通过yield逐步产出结果消费者用一个async for就能拿到值。它解决的核心问题是什么我举一个特别典型的场景边下载边处理。比如你要从一个接口拉取大量数据一次性全部拉回来内存扛不住或者响应时间太长体验很差。这时候用异步生成器每拿到一批数据就yield给上层处理上层可以边消费边准备下一步请求。用同步生成器做这种事得先同步地等每一次网络响应无法并发用普通异步函数做又没法分批次产出。异步生成器正好把这两者缝在一起。3.2 async for 的本质先看一个最简单的异步生成器async def countdown(n): while n 0: await asyncio.sleep(0.5) yield n n - 1 async def main(): async for num in countdown(3): print(num)async for语法上很像for但它内部做的事情完全不同。普通for循环迭代时调用的是__next__()同步地取下一个值async for每次迭代调用的是__anext__()这是一个异步方法返回的是一个可等待对象事件循环会等它完成再继续。换句话说async for是对“异步迭代器协议”的语法糖。手动实现它大概是这样的gen countdown(3) try: while True: num await anext(gen) print(num) except StopAsyncIteration: passanext()是 Python 3.10 加入的内置函数等价于普通生成器里的next()。在 3.10 之前你得gen.__anext__()然后await它丑得很。3.3 anext() 和异步生成器的手动迭代为什么需要手动迭代因为有些场景async for不够用。比如你要跳过前几个元素或者要根据条件提前终止迭代async for的固定模式做不到。# 跳过第一个元素 agen async_source() await anext(agen) # 丢弃第一个 async for item in agen: process(item)这种需求在实际数据流处理里挺常见的比如数据源的第一行往往是列名不是真正的数据。也要小心的一个细节异步生成器是冷启动的。创建它时不执行任何代码必须等第一次await anext()或async for第一次迭代才开始执行到第一个yield。这个特性和异步函数一样——“只声明不动手”。4. 实战一个带异步生成器的流式下载场景4.1 场景设计光讲概念不落地等于白看。我们设计一个真实场景有一个分页接口每页返回 100 条记录总共有几千页。我们要做的是用异步函数并发请求多个页面用异步生成器把结果逐条产出上层拿到每一条记录就立即处理比如入库为什么用异步生成器而不用列表收集因为几千页的数据全收进内存几十万条记录可能直接把内存打爆。流式处理的需求就是“来一条处理一条”内存占用恒定。4.2 完整代码示例import asyncio import aiohttp async def fetch_page(session, page: int): url fhttps://api.example.com/items?page{page} async with session.get(url) as resp: resp.raise_for_status() data await resp.json() return data.get(items, []) async def item_stream(session, max_page: int): 异步生成器按页拉取数据逐条产出 item for page in range(1, max_page 1): items await fetch_page(session, page) if not items: break for item in items: yield item async def main(): async with aiohttp.ClientSession() as session: # 并发拉取但处理时仍是逐条流式 sem asyncio.Semaphore(5) # 控制并发数别把服务器打挂 async def limited_fetch(page): async with sem: return await fetch_page(session, page) # 用 gather 并发拿多页数据然后逐条处理 pages await asyncio.gather( *(limited_fetch(p) for p in range(1, 11)) ) count 0 for page_items in pages: for item in page_items: # 这里做逐条处理比如写入数据库 count 1 print(fprocessed {count} items) # 如果数据量极大建议用异步生成器方式 async for item in item_stream(session, 10): # 每条 item 到达就立即处理 pass asyncio.run(main())4.3 关键细节解读这个例子看似简单里面藏了至少四个重要的设计决策第一ClientSession用async with管理。aiohttp 的 session 底层维护了连接池必须正确关闭否则会有连接泄漏风险。async with会保证会话结束后连接被释放这个不能省。第二用Semaphore控制并发度。并发不是越大越好。我实测过并发 50 个请求和并发 5 个请求总耗时差异不大瓶颈在目标服务器但并发过高会导致大量连接被重置甚至被封 IP。给并发加上限是对自己和目标服务器双方的尊重。第三gather和异步生成器的取舍。gather适合“并发等全部结果再统一处理”的场景优点是代码简单、好理解异步生成器适合“边拉边处理”的场景优点是内存占用低、管道流式传递。两者不是替代关系是不同场景下的不同工具。第四fetch_page里的await resp.json()不能省略。aiohttp 的响应对象拿到手时body 可能还没下载完。await resp.json()会等 body 全部完成再解析。这一点和 requests 不一样requests 是阻塞的响应返回时 body 已经完整了。5. 常见问题与排查技巧实录5.1 问题async 函数里直接写 time.sleep太经典了我自己早期也中过招。在异步函数里用time.sleep(1)整个事件循环会被阻塞 1 秒。事件循环是单线程的你阻塞了它所有其它任务全部暂停。# 错阻塞整个事件循环 async def bad_task(): time.sleep(1) # 事件循环卡死 1 秒 # 对让出控制权 async def good_task(): await asyncio.sleep(1) # 挂起当前任务循环可以跑别的排查方法程序总耗时和预期不符比自己算的并发时间慢好几倍先怀疑是不是有同步阻塞调用混在异步代码里。5.2 问题忘了 await返回的是 coroutineresult fetch_data(url) # result 是一个 coroutine 对象不是数据这可能是异步编程最常见的新手错误。忘了await变量拿到的是协程对象而不是函数返回值。更微妙的是如果这个协程对象没被任何地方 await 或调度Python 会抛RuntimeWarning: coroutine was never awaited。如果你用的是 PyCharm 这类 IDE它会用波浪线提示你别忽略它。排查方法看到coroutine was never awaited警告逐个检查所有调用异步函数的地方确认是否都有await。5.3 问题async for 里不小心用了同步大循环异步生成器本身没毛病但消费者如果写了一个耗时的同步循环同样会阻塞事件循环。async for item in item_stream(...): # 这里如果做重 CPU 计算比如解析超大 JSON、压缩图片 # 事件循环一样被卡住 heavy_compute(item)遇到这种情况要么把重计算丢给线程池asyncio.to_thread要么就根本不该用 asyncio直接上多进程。这点我在后面还会展开。排查方法事件循环卡顿时用py-spy dump --pid PID看当前线程堆栈如果栈停在某个同步循环里基本就实锤了。5.4 问题异步生成器不自动执行必须被消费异步生成器不同async for或anext驱动函数体永远不执行。很多人写了个异步生成器在main里调用了一下发现什么都没有怀疑代码有问题。实际上只是没消费。gen asyncio.gather(...) # 不跑 agen item_stream(session, 10) # 不跑 async for x in agen: # 跑起来了 pass记住创建异步生成器对象是零成本操作真正的执行发生在第一次迭代时。这个特性在调试时很头疼因为代码不执行到那个位置你根本看不出来有没有 bug。5.5 排查工具与技巧调试异步程序比调试同步程序难度高一个量级因为调用栈不连续任务切换会把你的思维打断。我的经验是第一个基础技巧是在关键位置打印日志并且带上任务标识。用asyncio.current_task()拿到当前任务对象打印出任务的名字或 id这样看日志时能分清哪一段属于哪个任务。import asyncio async def task_with_log(name): task asyncio.current_task() print(f[{name}] task_id{id(task)} start) await asyncio.sleep(1) print(f[{name}] task_id{id(task)} end)第二用asyncio.run(main(), debugTrue)开启调试模式。它会提醒你哪些协程没 await、哪些调用耗时超过 100ms默认阈值、是否发生事件循环阻塞。对于初学者排查“代码没反应”这类问题非常有帮助。第三遇到诡异问题别瞎猜用asyncio.all_tasks()打印出所有待处理任务看看是不是有任务卡住了或者泄漏了。pending [t for t in asyncio.all_tasks() if not t.done()] for t in pending: print(pending task:, t)第四个技巧如果一个任务怎么都找不到退出条件试试给wait_for包一层超时让它强制失败。这既是排查手段也是保险措施。try: await asyncio.wait_for(suspicious_task(), timeout3) except asyncio.TimeoutError: print(任务超时强制退出问题在任务内部)6. 一段关于性能边界的经验之谈6.1 异步不是银弹我得说点泼冷水的话。异步编程最大的好处是让 I/O 密集型任务的吞吐量飙升但它不是万能的。如果任务是 CPU 密集型的比如加密解密、图像处理、复杂数值计算asyncio 不仅帮不上忙反而因为 Python GIL 的限制多任务之间几乎没法真正并行。你写一堆async def跑密集计算事件循环照样得一个一个算白白增加调度开销。一个简单的判定方法你的任务里是“等数据”的时间多还是“算数据”的时间多如果 90% 的时间在等网络响应、等数据库返回、等磁盘 I/O异步是正确选择如果 90% 的时间在算东西异步没什么意义应该用concurrent.futures.ProcessPoolExecutor或直接上多进程。6.2 什么时候该用多线程/多进程用 asyncio 的asyncio.run_coroutine_threadsafe和asyncio.to_thread可以在异步程序里桥接多线程# 在异步代码里跑一个阻塞函数丢到线程池去执行 result await asyncio.to_thread(requests.get, https://example.com)to_thread是实现“同步库适配异步框架”的经典技巧。比如你还在用 requests 但想用 async 框架FastAPI你不需要把所有代码改成 aiohttp直接把 requests 调用丢给线程池就行。它会在线程池里跑阻塞的只是池里的线程不会卡事件循环。多进程的话ProcessPoolExecutor配合asyncio用是有的但复杂度很高数据要 pickle 序列化、进程间通信麻烦、调试困难非必要不推荐。6.3 我的取舍经验最后说说我自己做技术选型时的经验。单机搞并发我会先问三个问题是 I/O 密集还是 CPU 密集前者选异步后者选多进程。现有代码库大量使用了同步库吗如果是优先考虑to_thread而不是全量重写。团队熟悉异步吗异步代码写了一段时间你会发现它维护起来比同步代码难得多特别是并发 bug、竞态条件和难看的调用栈。如果团队没有异步经验可能用线程池 同步代码反而更靠谱。异步使吞吐量上来了但代码可读性和心智负担也上来了。这个 trade-off只有上过生产环境的人才有体会。网上那些“一招教你高并发爬虫”的教程从来不提这个但我见过太多人用 asyncio 把代码写得一团糟最后性能没上去多少反而 bug 翻倍。再补一个我踩过的真实坑在异步代码里如果用了requests这类同步库遇到网络超时会非常恶心。requests 默认没有超时卡住的话整个任务一直挂着事件循环里其他任务好好的但这个任务永远结束不了。你在wait_for里设置了超时它能取消 await但取消不掉底层线程池里的实际请求——那个线程还在等响应。这是异步 I/O 和同步 I/O 混合使用时最大的暗坑。我的建议是一旦决定走全异步路线所有网络调用必须用 aiohttp 或 httpx 的异步模式不要混用同步库否则出了问题排查成本极高。异步函数和异步生成器的核心模型其实不复杂复杂的是在真实场景里怎么设计任务边界、怎么控制并发、怎么处理取消和超时。这篇文章没有覆盖所有细节比如asyncio.Queue的生产者消费者模式、async with配合异步上下文管理器实现资源自动回收这些内容任何一个单独拿出来都能再写一篇长文。把握住一个主线就够用了异步让 I/O 等待不再阻塞其他任务异步生成器让数据流可以按需加工、按需消费两者叠加无论处理多大规模的数据内存和吞吐量都能在可控范围。剩下的细节在实战里慢慢体会就行。