Python异步编程从入门到实战：结合RAG流式回答全解析

一、前言在Python开发中高并发场景如API服务、网络爬虫、大模型调用、RAG应用越来越常见而异步编程正是解决这类场景的核心技术。相比于传统同步编程异步编程能极大提升IO密集型任务的执行效率避免CPU资源浪费相比于多线程异步编程开销更低、并发能力更强尤其适合高并发场景下的开发。前置要求Python 3.7异步简化写法asyncio.run()为3.7新增低于该版本需升级终端输入python --version可查看版本。二、异步编程核心基础新手学习异步最容易困惑的是“异步到底是什么”“为什么比同步快”。我们先从同步与异步的对比入手用生活场景代码示例直观理解两者的差异再逐步过渡到异步的核心逻辑。2.1 同步vs异步生活例子核心区别先抛开复杂的技术概念用生活中最常见的“泡茶”任务理解同步与异步的本质区别。2.1.1 生活场景类比假设我们有3个任务需要完成烧开水2分钟、洗杯子1分钟、泡茶0.5分钟两种执行方式如下同步执行串行烧水壶加水 → 傻等水开2分钟期间什么都不做 → 洗杯子1分钟 → 泡茶0.5分钟。总耗时210.53.5分钟等待期间CPU人完全闲置。异步执行并发烧水壶加水 → 不傻等水开立刻去洗杯子1分钟 → 洗杯子期间水开了剩余1分钟继续洗杯子 → 洗完杯子立刻泡茶0.5分钟。总耗时20.52.5分钟等待期间CPU人不闲置高效利用时间。这个例子的核心的是同步是“傻等”异步是“不等”——等待某个任务如烧水的同时去执行其他可执行的任务从而提升整体效率。2.1.2 核心区别对照表特性同步Synchronous异步Asynchronous执行方式任务按顺序执行上一个任务完成后下一个才开始任务并发执行某个任务等待时切换到其他任务执行资源利用等待期间如IO操作CPU完全闲置资源浪费严重等待期间CPU去执行其他任务资源利用率极高底层逻辑单线程串行无需额外调度单线程事件循环主动切换任务适用场景简单小任务、无IO等待如简单数值计算IO密集型任务网络请求、文件读写、大模型调用、数据库操作代码复杂度简单逻辑直观稍复杂需掌握异步语法async/await等2.2 同步vs异步代码对比我们用“模拟3个IO任务每个任务等待一定时间”的场景写同步和异步代码直观感受两者的效率差异代码逐行添加注释新手可直接复制运行理解每一步的作用。2.2.1 同步代码基础对比理解“傻等”同步代码的核心是“串行执行”每个任务必须等上一个任务完成才能开始总耗时是所有任务耗时之和。# 1. 导入时间库用于计时和模拟IO等待如网络请求、文件读写 import time # 2. 定义同步任务函数模拟IO等待操作 # 参数说明name任务名称delay等待时间秒模拟IO操作的耗时 def sync_task(name, delay): print(f【同步】任务{name}开始等待{delay}秒模拟IO操作) # 核心time.sleep(delay) 是同步阻塞等待——程序啥也不干就傻等delay秒 # 此时CPU完全闲置无法执行其他任务 time.sleep(delay) print(f【同步】任务{name}等待完成继续执行下一个任务\n) # 3. 程序入口Python规范写法保证代码只在直接运行时执行避免导入时执行 if __name__ __main__: # 记录程序开始时间用于计算总耗时 start_time time.time() # 4. 按顺序执行3个同步任务串行执行必须等上一个完成 sync_task(A, 2) # 任务A等待2秒 sync_task(B, 1) # 任务B等待1秒必须等A完成才能开始 sync_task(C, 3) # 任务C等待3秒必须等B完成才能开始 # 5. 计算并打印总耗时保留2位小数更直观 total_time time.time() - start_time print(f【同步】所有任务执行完成总耗时{total_time:.2f}秒)执行结果完全串行总耗时2136秒【同步】任务A开始等待2秒模拟IO操作 【同步】任务A等待完成继续执行下一个任务 【同步】任务B开始等待1秒模拟IO操作 【同步】任务B等待完成继续执行下一个任务 【同步】任务C开始等待3秒模拟IO操作 【同步】任务C等待完成继续执行下一个任务 【同步】所有任务执行完成总耗时6.00秒关键总结同步代码的问题在于“等待期间CPU闲置”哪怕有其他任务可以执行也必须傻等上一个任务完成效率极低。2.2.2 异步代码核心实现理解“不等”我们将上面的同步代码改写为异步仅新增4个核心异步语法就能实现并发执行总耗时缩短为最长任务的耗时3秒效率直接翻倍。# 1. 导入异步编程核心库Python内置无需额外安装 # asyncio是异步编程的核心提供事件循环、协程管理等功能 import asyncio # 导入时间库用于计算总耗时和同步代码一致 import time # 2. 定义异步任务函数用async def替代普通def标记为异步函数 # 异步函数的核心特点可以使用await语法实现非阻塞等待 async def async_task(name, delay): print(f【异步】任务{name}开始等待{delay}秒模拟IO操作) # 核心await asyncio.sleep(delay) 是异步非阻塞等待 # 含义暂停当前任务主动让出CPU让事件循环去执行其他任务 # 等delay秒后事件循环再回来继续执行当前任务 await asyncio.sleep(delay) print(f【异步】任务{name}等待完成\n) # 3. 定义主异步函数用于管理所有异步任务 # 注意异步任务不能直接在程序入口执行必须通过主异步函数管理 async def main(): # 3.1 创建异步任务列表将3个异步任务封装成可并发执行的Task # asyncio.create_task()将异步函数转为Task加入事件循环等待执行 # 此时任务已被事件循环接管会在合适的时机执行无需等待上一个任务 task1 asyncio.create_task(async_task(A, 2)) task2 asyncio.create_task(async_task(B, 1)) task3 asyncio.create_task(async_task(C, 3)) # 3.2 等待所有异步任务完成 # asyncio.gather(*tasks)等待列表中的所有Task完成*tasks表示拆分成单个参数 # 这里会暂停main函数直到所有任务都执行完毕 await asyncio.gather(task1, task2, task3) # 4. 程序入口启动异步程序 if __name__ __main__: start_time time.time() # asyncio.run(main())异步程序的“总开关” # 作用自动创建事件循环、执行main主异步函数、执行完成后关闭事件循环 # 无需手动管理事件循环简化异步编程流程Python 3.7新增 asyncio.run(main()) # 计算总耗时 total_time time.time() - start_time print(f【异步】所有任务执行完成总耗时{total_time:.2f}秒)执行结果并发执行总耗时≈3秒即最长任务的耗时【异步】任务A开始等待2秒模拟IO操作 【异步】任务B开始等待1秒模拟IO操作 【异步】任务C开始等待3秒模拟IO操作 【异步】任务B等待完成 【异步】任务A等待完成 【异步】任务C等待完成 【异步】所有任务执行完成总耗时3.00秒关键总结异步代码的核心是“非阻塞等待”——当某个任务遇到await需要等待时会主动让出CPU让其他任务执行从而实现并发大幅提升效率。2.2.3 异步代码执行流程拆解新手必懂很多新手虽然能运行异步代码但不理解底层执行流程这里用时间轴拆解上面的异步代码清晰看到“任务切换”的过程0秒asyncio.run(main())启动事件循环执行main函数创建task1、task2、task3三个任务事件循环依次启动这三个任务。 0秒紧接着执行task1任务A打印“开始等待2秒”遇到await asyncio.sleep(2)暂停task1主动让出CPU事件循环切换到task2。 0秒紧接着执行task2任务B打印“开始等待1秒”遇到await asyncio.sleep(1)暂停task2主动让出CPU事件循环切换到task3。 0秒紧接着执行task3任务C打印“开始等待3秒”遇到await asyncio.sleep(3)暂停task3此时所有任务都处于等待状态事件循环暂时“休息”。 1秒task2的等待时间1秒到了事件循环唤醒task2执行剩余代码打印“任务B等待完成”task2执行完毕。 2秒task1的等待时间2秒到了事件循环唤醒task1执行剩余代码打印“任务A等待完成”task1执行完毕。 3秒task3的等待时间3秒到了事件循环唤醒task3执行剩余代码打印“任务C等待完成”task3执行完毕。 3秒紧接着asyncio.gather()检测到所有任务完成main函数执行完毕事件循环关闭程序结束。从流程可以看出异步的“并发”本质是“任务切换”——等待时让出CPU避免闲置从而提升整体效率。三、核心语法深度解析新手必懂避坑重点异步编程的核心语法只有4个async def、await、yield、async for再加上实战必备的锁机制with db_lock。本章逐字逐句拆解每个语法的作用、用法、常见错误结合示例讲解确保新手能彻底理解避免踩坑。3.1 async def异步函数的“身份证”3.1.1 语法格式async def 函数名(参数列表): # 函数体可包含await、yield等异步语法 pass3.1.2 核心作用标记一个函数为异步函数告诉Python解释器这个函数可以使用await、yield等异步语法执行时不会阻塞事件循环能主动让出CPU。3.1.3 关键注意事项避坑重点必须用async def定义只要函数内部需要使用await或yield就必须用async def定义否则会报错SyntaxError。异步函数调用不会直接执行普通函数调用如func()会直接执行并返回结果而异步函数调用如async_func()会返回一个“协程对象”不会直接执行必须通过事件循环如asyncio.run()或await触发执行。示例对比正确vs错误# 正确用async def定义异步函数可使用await async def correct_async_func(): await asyncio.sleep(1) print(异步函数执行完成) # 错误1普通def函数中使用await报错 def wrong_sync_func1(): await asyncio.sleep(1) # 报错SyntaxError: await outside async function # 错误2直接调用异步函数不会执行 async def async_func(): print(异步函数执行) async_func() # 仅返回协程对象不会打印任何内容需用await或asyncio.run()触发3.2 await非阻塞等待的“核心开关”await是异步编程的“灵魂”核心作用是“暂停当前协程让出CPU等待某个操作完成后再恢复执行”也是实现“不傻等”的关键。3.2.1 语法格式await 可等待对象3.2.2 什么是“可等待对象”await后面必须跟“可等待对象”否则会报错常见的可等待对象有3种异步函数async def定义的函数返回的协程对象asyncio.sleep()、asyncio.gather()等asyncio库提供的异步函数异步迭代器如LLM的astream()返回的对象、Future对象。3.2.3 核心作用拆解结合生活例子还是用“泡茶”场景await就相当于“烧水壶加水后不傻等去洗杯子”——暂停当前任务等水开让出CPU去洗杯子等水开了等待完成再回来继续泡茶恢复当前任务。代码层面await的作用是暂停当前异步函数的执行将CPU资源让给事件循环让事件循环去执行其他可执行的异步任务等待await后面的“可等待对象”执行完成可等待对象执行完成后恢复当前异步函数的执行继续执行await后面的代码。3.2.4 常见错误与避坑技巧错误1await用在普通def函数中如3.1.3中的错误示例会直接报错必须用async def定义函数。错误2用time.sleep()替代asyncio.sleep()time.sleep()是同步阻塞函数哪怕放在async def函数中也会卡住整个事件循环导致异步失效总耗时和同步一样。错误3await后面跟普通函数普通函数不是“可等待对象”用await修饰会报错。import asyncio import time async def await_demo(): # 正确await 可等待对象asyncio.sleep是异步函数 await asyncio.sleep(2) print(异步等待完成) # 错误2用time.sleep()同步阻塞卡住事件循环 # time.sleep(2) # 错误3await 普通函数非可等待对象 # def normal_func(): # return 1 # await normal_func() # 报错TypeError: object int cant be used in await expression asyncio.run(await_demo())3.3 yield异步生成器的“逐段返回”神器yield是实现“流式返回”的核心语法尤其适用于大模型流式回答、日志逐行输出、大数据分批处理等场景——函数不是一次性返回所有结果而是“生成一个、返回一个”像流水线发货一样逐段返回数据。结合async defyield就形成了“异步生成器”既能实现异步非阻塞又能逐段返回数据是RAG流式回答、大模型API调用的核心技术。3.3.1 先懂普通yield基础铺垫在学习异步生成器之前先了解普通生成器def yield理解yield的核心逻辑——逐段返回数据函数暂停执行。# 普通生成器函数def yield逐段返回数据 def generate_data(): # 第一次遍历返回第一段数据函数暂停在yield处 yield 第一段数据 # 第二次遍历从暂停处继续执行返回第二段数据函数暂停 yield 第二段数据 # 第三次遍历从暂停处继续执行返回第三段数据函数结束 yield 第三段数据 # 遍历生成器用普通for循环拿一个、生成一个 for data in generate_data(): print(data)执行结果逐段输出第一段数据 第二段数据 第三段数据核心特点yield不是“结束函数”而是“临时返回一个值函数暂停在当前位置”每次遍历生成器for循环函数从暂停处继续执行直到下一个yield如果没有更多yield函数结束生成器迭代完成。3.3.2 异步生成器async def yield实战核心异步生成器是“异步函数”和“生成器”的结合语法是async def yield核心作用是在异步非阻塞的基础上逐段返回流式数据。3.3.2.1 语法格式async def 异步生成器函数(参数列表): # 函数体 yield 数据片段 # 逐段返回数据 await 可等待对象 # 可结合异步等待实现非阻塞流式返回3.3.2.2 核心应用场景最常见的场景是“大模型流式回答”大模型生成回答时不是一次性返回完整文本而是先生成一个字、一个词再逐段返回前端拿到一个片段就显示一个片段实现“打字机”效果提升用户体验。3.3.2.3 代码示例模拟大模型流式回答import asyncio # 异步生成器函数模拟大模型流式回答逐段返回文本 async def llm_stream_answer(prompt): # 模拟大模型生成的文本片段实际场景中是从大模型API获取的流式片段 answer_chunks [你, 好, , 我, 是, 异, 步, 生, 成, 器] for chunk in answer_chunks: yield chunk # 逐段返回文本片段 # 模拟打字机效果每个片段返回后等待0.02秒再返回下一个 await asyncio.sleep(0.02) # 调用异步生成器必须用async for遍历普通for会报错 async def main(): # async for 遍历异步生成器逐段获取数据 async for chunk in llm_stream_answer(你好): # end不换行flushTrue实时打印避免缓存 print(chunk, end, flushTrue) # 启动异步程序 asyncio.run(main())执行效果文本逐字输出像打字机一样而非一次性打印全部内容你好我是异步生成器。3.3.3 yield vs return核心区别避坑重点很多新手会混淆yield和return两者的核心区别的是“一次性返回”和“逐段返回”语法核心作用函数状态适用场景return一次性返回所有结果返回后函数直接结束无法继续执行普通函数、无需流式返回的场景yield逐段返回数据每次返回一个片段返回后函数暂停下次遍历从暂停处继续执行流式返回大模型回答、日志输出、分批处理数据示例对比import asyncio # 用return一次性返回所有结果函数结束 async def return_demo(): return [你, 好, ] # 用yield逐段返回函数暂停 async def yield_demo(): yield 你 yield 好 yield async def main(): # 调用return的异步函数一次性获取所有结果 return_result await return_demo() print(return返回, return_result) # 输出return返回[你, 好, ] # 调用yield的异步生成器逐段获取结果 print(yield返回, end) async for chunk in yield_demo(): print(chunk, end, flushTrue) # 输出yield返回你好 asyncio.run(main())3.4 async for异步迭代器的“专属遍历工具”async for是专门用于遍历“异步迭代器”的语法比如异步生成器、大模型的astream()返回的对象、异步文件读取对象等。普通for循环无法遍历异步迭代器必须用async for否则会报错。3.4.1 核心原因异步迭代器的“迭代过程”是异步的比如获取下一个片段需要等待IO操作如大模型返回数据普通for循环是同步的无法处理异步等待会卡住整个程序因此必须用async for——异步遍历等待下一个片段时主动让出CPU不阻塞事件循环。3.4.2 语法格式async for 变量 in 异步迭代器: # 处理每一个迭代的片段 pass3.4.3 代码示例结合大模型流式调用实际开发中大模型的流式接口如llm.astream()返回的是异步迭代器必须用async for遍历逐段获取生成的文本import asyncio # 模拟大模型流式接口返回异步迭代器实际场景中无需自己实现调用大模型API即可 async def llm_astream(prompt): answer 异步编程是Python处理高并发IO场景的核心技术适合大模型流式调用、网络爬虫等场景。 # 逐字拆分文本模拟流式返回 for char in answer: yield char await asyncio.sleep(0.01) # 模拟大模型生成延迟 # 异步遍历大模型流式结果 async def main(): prompt 什么是异步编程 print(f提问{prompt}) print(回答, end, flushTrue) # async for 遍历异步迭代器逐段获取大模型生成的文本 async for chunk in llm_astream(prompt): print(chunk, end, flushTrue) asyncio.run(main())执行效果大模型的回答逐字输出实现“打字机”效果提升用户体验。常见错误用普通for遍历异步迭代器会报错# 错误用普通for遍历异步迭代器 async def main(): for chunk in llm_astream(什么是异步编程): # 报错TypeError: async generator object is not iterable print(chunk)3.5 with db_lock高并发下的共享资源保护神在高并发异步场景中多个任务同时操作“共享资源”如向量数据库、文件、全局变量时会出现“数据竞争”问题导致数据重复、损坏、报错甚至程序崩溃。此时就需要用互斥锁db_lock来保护共享资源而with db_lock是最安全、最简洁的加锁方式。3.5.1 核心概念db_lock是什么db_lock是一个互斥锁Mutex也叫排他锁你可以把它理解成“一个独占许可证”同一时间只有一个任务能拿到这个许可证加锁成功其他任务必须等待许可证归还解锁后才能拿到许可证执行相关操作。而with db_lock是Python的“上下文管理器”写法等价于手动加锁解锁但更安全避免忘记解锁导致死锁# 手动加锁/解锁不推荐容易忘解锁导致死锁 db_lock.acquire() # 获取锁拿许可证 try: # 操作共享资源如向量数据库写入 vector_db.add_texts(textsvalid_texts) finally: db_lock.release() # 释放锁还许可证无论是否报错都会执行 # 用with db_lock推荐自动加锁/解锁 with db_lock: # 操作共享资源执行完自动解锁无需手动处理 vector_db.add_texts(textsvalid_texts)3.5.2 核心作用解决数据竞争问题代码中vector_db.add_texts(textsvalid_texts)是操作“共享资源”向量数据库——比如多个用户同时调用generate_streaming_answer函数都会执行这行代码往同一个向量数据库里写入文本。如果不加锁多个任务会“抢着”写入数据导致各种异常如果加锁同一时间只有一个任务能写入保证数据安全。3.5.3 加与不加db_lock的具体影响场景多用户同时调用函数往向量数据库写入文本状态具体影响典型问题不加db_lock多个任务同时写入出现数据竞争数据不安全1. 数据重复插入同一文本多次写入2. 数据损坏一个任务写入一半另一个任务覆盖3. 数据库报错锁冲突、文件被占用4. 极端情况数据库崩溃加db_lock同一时间只有一个任务能写入数据安全但写入操作串行执行写入效率轻微下降合理代价但保证数据完整一致避免报错3.5.4 异步场景下为什么还需要锁很多新手会问“异步是单线程为什么还要加锁” 核心原因有3点共享资源底层可能是多线程/多进程你的代码是单线程异步但向量数据库如Chroma、FAISS、文件系统的底层实现可能用了多线程写入操作依然有并发风险。异步任务切换导致的并发写入异步事件循环中多个任务会在await处切换——比如任务A在执行vector_db.add_texts()时遇到await让出CPU任务B接着执行写入操作导致两个任务同时写入。程序扩展需求如果后续你的程序改成“多进程异步”比如用多进程提升并发能力锁依然能保护跨进程的共享资源避免后续修改代码时出现问题。3.5.5 锁的“粒度”技巧平衡安全与效率代码中锁只包裹了vector_db.add_texts(...)这一小段代码而不是整个函数——这是“最小锁粒度”原则也是实战中的核心技巧只锁“必须串行执行”的代码即操作共享资源的代码其他代码如文本过滤、检索、LLM调用依然可以并发执行目的平衡“数据安全”和“并发性能”避免整个函数串行执行那样异步就失去了意义。示例正确的锁粒度async def generate_streaming_answer(question, knowledge_texts, similarity_threshold): try: # 1. 文本过滤无需加锁可并发执行 if knowledge_texts: valid_texts [text.strip() for text in knowledge_texts if text.strip()] if valid_texts: # 2. 操作共享资源加锁串行执行 with db_lock: vector_db.add_texts(textsvalid_texts) # 3. 检索、LLM调用无需加锁可并发执行 retrieved_results retrieve_with_score(question, k5) # ... 后续逻辑 except Exception as e: yield f[错误] {str(e)}四、异步vs多线程核心差异与适用场景新手最容易混淆“异步”和“多线程”——两者都能处理并发但核心原理、开销、适用场景完全不同。很多人误以为“异步就是多线程”其实两者有本质区别本章用生活例子、代码对比、维度表格彻底讲透两者的差异帮助大家选对技术方案。4.1 核心维度对比表特性异步async/await多线程threading核心原理单线程 事件循环协程主动让出CPU协作式并发多线程 操作系统调度线程被动切换抢占式并发调度方式协作式只有协程遇到await时才主动让出CPU程序员可控制切换时机抢占式操作系统随时可能暂停一个线程切换到另一个线程程序员无法控制切换开销极低程序内部函数调用级别的切换无系统调用资源占用可忽略协程仅保存函数上下文较高操作系统内核态切换需要消耗额外资源线程栈默认几MB还有内核调度成本并发能力极强可支撑百万级协程并发如同时处理10万用户请求较弱最多支撑几千个线程再多会耗尽内存/CPU导致程序卡死GIL影响Python无影响全程单线程GIL一直被占用不影响异步执行有严重影响CPU密集型任务无法真正并行同一时间只有1个线程执行Python字节码仅IO密集型任务能受益线程安全无需考虑单线程执行不存在多任务同时修改共享资源的问题除了底层多线程实现的共享资源需要考虑多线程同时修改共享资源时会出现数据竞争必须加锁保护编程难度稍高需掌握async/await/yield等异步语法理解事件循环较低语法接近同步代码新手容易上手适用场景高并发IO密集型大模型流式调用、API服务、网络爬虫、实时聊天、数据库操作低/中并发IO密集型简单文件下载、少量数据库查询、老项目兼容同步库4.2 生活例子餐厅服务模型直观理解差异用“餐厅服务顾客”的场景类比异步和多线程的差异更容易理解假设餐厅有3个顾客3个任务每个顾客需要“点餐→等餐→上菜”IO操作等后厨做餐耗时但不用服务员盯。4.2.1 多线程雇多个服务员多线程餐厅雇了3个服务员3个线程每个服务员盯1个顾客1个任务顾客A点餐→服务员1把订单给后厨→站在原地傻等线程阻塞直到后厨喊“餐好了”再给A上菜同时服务员2、3分别盯顾客B、C流程一样问题服务员线程数量有限比如最多雇10个雇多了老板操作系统管理成本高线程切换开销且服务员傻等时完全闲置。4.2.2 异步1个全能服务员事件循环协程餐厅只雇1个全能服务员事件循环协程同时盯3个顾客顾客A点餐→服务员把订单给后厨→不傻等立刻去服务顾客B协程await时让出CPU后厨喊“顾客A的餐好了”→服务员暂停服务B回去给A上菜→上完再继续服务B核心服务员全程不闲着1个人干3个人的活且管理成本极低不用雇多人。4.3 代码对比IO任务的两种实现方式用“模拟3个IO任务每个等待2秒”的场景分别用多线程和异步实现直观感受两者的效率和代码差异。4.3.1 多线程版本threadingimport threading import time # 定义任务函数每个线程执行这个函数 def thread_task(name): print(f【多线程】任务{name}开始等待2秒模拟IO操作) time.sleep(2) # 同步阻塞等待线程傻等占着资源 print(f【多线程】任务{name}等待完成\n) if __name__ __main__: start_time time.time() # 创建3个线程每个线程处理1个任务 threads [ threading.Thread(targetthread_task, args(A,)), threading.Thread(targetthread_task, args(B,)), threading.Thread(targetthread_task, args(C,)) ] # 启动所有线程 for t in threads: t.start() # 等待所有线程结束join()阻塞主线程直到子线程全部完成 for t in threads: t.join() print(f【多线程】总耗时{time.time()-start_time:.2f}秒)执行结果总耗时≈2秒和异步一样但底层逻辑不同【多线程】任务A开始等待2秒模拟IO操作 【多线程】任务B开始等待2秒模拟IO操作 【多线程】任务C开始等待2秒模拟IO操作 【多线程】任务A等待完成 【多线程】任务B等待完成 【多线程】任务C等待完成 【多线程】总耗时2.00秒核心问题3个线程在time.sleep(2)时都是“阻塞状态”占着系统资源线程栈、内核调度表如果开1000个线程系统开销会急剧上升甚至卡死。4.3.2 异步版本async/awaitimport asyncio import time async def async_task(name): print(f【异步】任务{name}开始等待2秒模拟IO操作) await asyncio.sleep(2) # 异步非阻塞等待主动让出CPU print(f【异步】任务{name}等待完成\n) async def main(): start_time time.time() tasks [ asyncio.create_task(async_task(A)), asyncio.create_task(async_task(B)), asyncio.create_task(async_task(C)) ] # 等待所有异步任务完成 await asyncio.gather(*tasks) # 计算并打印总耗时 total_time time.time() - start_time print(f【异步】总耗时{total_time:.2f}秒) # 程序入口启动异步程序 if __name__ __main__: asyncio.run(main())执行结果总耗时≈2秒与多线程耗时一致但资源开销更低【异步】任务A开始等待2秒模拟IO操作 【异步】任务B开始等待2秒模拟IO操作 【异步】任务C开始等待2秒模拟IO操作 【异步】任务A等待完成 【异步】任务B等待完成 【异步】任务C等待完成 【异步】总耗时2.00秒核心优势同样实现2秒并发完成3个IO任务但异步仅用1个线程无线程切换开销资源占用极低即使扩展到1000个任务也能稳定运行不会出现多线程的内存耗尽问题。