对象是组织代码的方式并发是组织执行的方式——这是 Python 最值得搞懂的两块。而且Python 并发绕不开一个东西GIL一、面向对象1.1 先搞懂类和对象是什么类是图纸对象是按图纸造出来的实物。一张账户图纸可以造出无数个账户每个账户有自己的主人数据。classAccount:def__init__(self,owner,balance0):self.ownerowner# 这个对象自己的属性self.balancebalance accAccount(小明,100)# 造出一个对象print(acc.owner)# 小明print(acc.balance)# 100三个初学必懂的点__init__是构造函数Account(小明, 100)执行时自动调用负责把新对象初始化好。self就是这个对象自己acc.deposit(50)其实等于Account.deposit(acc, 50)——Python 自动把acc填进第一个参数self。为什么用类把数据 操作这些数据的函数捆在一起账户的余额只该由账户的方法来改而不是散落在代码各处谁都能acc.balance -999。1.2 让对象能被 print__repr__直接 print 对象会得到__main__.Account object at 0x...调试时看不出内容。定义__repr__就能自定义打印的样子classAccount:def__init__(self,owner,balance0):self.ownerowner self._balancebalancedef__repr__(self):returnfAccount({self.owner!r},{self._balance})print(Account(小明,100))# Account(小明, 100)1.3 属性保护与 propertyPython 没有真正的 private私有变量靠约定单下划线_balance表示内部使用请别从外面直接改。但约定拦不住人真想要写入时校验用propertyclassAccount:def__init__(self,owner,balance0):self.ownerowner self._balancebalancepropertydefbalance(self):# 读: x acc.balance 走这里returnself._balancebalance.setterdefbalance(self,value):# 写: acc.balance x 走这里ifvalue0:raiseValueError(余额不能为负)self._balancevaluedefdeposit(self,amount):ifamount0:raiseValueError(存款必须为正)self._balanceamountreturnself._balance accAccount(小明,100)acc.balance-1# 触发 setter, 抛出: 余额不能为负顺带解释装饰器Python 里写在函数/方法头顶上、以开头的那行东西叫装饰器作用是给函数套一层壳再交还给你。property的壳做的事是把这个方法伪装成属性——外面用acc.balance读写不加括号实际却在偷偷走你写的方法。为什么值得这么做调用方始终写acc.balance将来要加校验只改 setter 一处所有调用代码一行不动——这就是 property 存在的理由。1.4 三种方法的分工classAccount:# ...前面的 __init__ / property / deposit 略...classmethoddefzero(cls,owner):替代构造器: 常用于从配置/数据库行创建对象returncls(owner,balance0)# cls 就是这个类自己staticmethoddefis_valid_amount(amount):纯工具函数, 跟实例/类状态都无关returnisinstance(amount,(int,float))andamount0Account.zero(新用户)# 不用先有实例, 直接用类调用Account.is_valid_amount(50)# True方法第一个参数用途普通方法self操作某个对象的状态绝大多数情况classmethodcls替代构造器、工厂方法staticmethod无纯工具函数1.5 继承、super、多态继承子类复用父类的全部代码再加自己的东西。classSavingsAccount(Account):# 括号里写父类, 就完成了继承def__init__(self,owner,balance0,rate0.02):super().__init__(owner,balance)# super() 调用父类的同名方法self.raterate# 子类新增的属性defadd_interest(self):# 子类新增的方法self._balance*(1self.rate)returnself._balancedefmonthly_statement(acc):多态: 只用 acc 的属性和方法, 不关心它具体是哪种账户returnf{acc.owner}:{acc.balance:.2f}sSavingsAccount(小明,1000,rate0.03)s.add_interest()print(monthly_statement(s))# 对 Account 的子类同样适用monthly_statement收的是Account传SavingsAccount进去照样工作——多态在 Python 里是免费的不用像 Java 那样先声明接口函数里只调了acc.owner和acc.balance任何有这两个属性的对象都能传进来。1.6 运算符重载让自定义对象像内置类型一样用1 2、len([1,2])、print(obj)这些写法背后其实都在调用对象身上的特殊方法名字都是双下划线包裹俗称魔法方法。自己实现了它们自己的对象就获得同样的能力classMoney:def__init__(self,amount,currency):self.amountamount self.currencycurrencydef__repr__(self):# print(m) 时的样子returnfMoney({self.amount}{self.currency})def__add__(self,other):# 让 Money 支持 号ifself.currency!other.currency:raiseValueError(币种不同不能直接相加)returnMoney(self.amountother.amount,self.currency)def__lt__(self,other):# 让 Money 支持 和 比较大小returnself.amountother.amountdef__bool__(self):# 定义 if m: 的行为returnself.amount!0a,bMoney(100,CNY),Money(58,CNY)print(ab)# Money(158 CNY)print(ab)# Trueprint(bool(Money(0,CNY)))# False同理实现__len__就能被len()量、实现__iter__就能被for遍历、实现__enter__/__exit__就能配合with使用。能干什么取决于实现了哪些方法。二、并发模型先直面 GIL2.1 先搞懂进程和线程是什么用一个厨房打比方程序 一份菜谱躺在硬盘上的代码进程 一间厨房程序跑起来后操作系统分给它独立的锅碗瓢盆——内存空间线程 厨房里的厨师每间厨房至少一个厨师同一厨房的厨师共享锅碗瓢盆CPU 核心 灶台的数量两个关键概念并行灶台不止一个多个厨师真的同时开火炒菜。并发灶台只有一个但厨师 A 炖上汤去等的时候厨师 B 插上来炒自己的菜——交替使用看起来像同时。2.2 GIL 是什么——用实验说话先补一句背景Python 代码不是直接在 CPU 上运行的而是先被翻译成字节码交给解释器执行。CPython最常用的官方解释器有个规定GILGlobal Interpreter Lock全局解释器锁同一时刻只允许一个线程执行 Python 字节码。用厨房的话说这间厨房不管有多少厨师灶台使用许可证只有一张——同一时刻只有一个厨师能真正开火其他人只能干等。空口无凭做实验。用同一个纯计算函数跑三遍串行 / 2 线程 / 2 进程importtimefrommultiprocessingimportPoolfromthreadingimportThread N80_000_000defcpu_bound(n):纯 CPU 计算: 一直占用灶台烧个不停的任务total0foriinrange(n):totali*ireturntotaldeftimeit(label,fn):t0time.perf_counter()fn()costtime.perf_counter()-t0print(f{label:12}{cost:.2f}s)if__name____main__:WORK[N,N]# 两个同样大小的任务# 1. 串行: 一个接一个做timeit(串行,lambda:[cpu_bound(n)forninWORK])# 2. 多线程: 同一进程内的两个厨师, 但许可证只有一张defrun_threads():ts[Thread(targetcpu_bound,args(n,))forninWORK]fortints:t.start()fortints:t.join()# join 等这个线程干完再往下走timeit(多线程,run_threads)# 3. 多进程: 再开一间厨房, 每间厨房有自己的许可证defrun_procs():withPool(2)asp:p.map(cpu_bound,WORK)timeit(多进程,run_procs)实测输出Python 3.13串行 10.51s 多线程 10.34s 多进程 6.29s三个结论多线程 10.34s ≈ 串行 10.51s几乎零加速。两张许可证不存在的——两个线程轮流持锁执行还多付了切换开销。多进程 6.29s——每个进程有独立的解释器和独立的 GIL两间厨房各有各的许可证真用上了 2 个核。没到理想的 5.25s 是因为开新厨房——进程创建和数据搬运——本身有开销。GIL 的准确表述它限制同一时刻执行 Python 字节码的线程数 1不是Python 不能多线程下面马上看到反例。2.3 GIL 什么时候会释放关键规则线程做 IO等网络、读写磁盘、time.sleep时会释放 GIL。回到厨房厨师把汤炖上、站在旁边等的时候等待不需要灶台许可证会让给别的厨师用。任务类型多线程有没有用原因IO 密集网络请求、读写文件✅ 有用等 IO 时释放 GIL其他线程干活CPU 密集数值计算、解析大 JSON❌ 没用字节码全程持锁退化成串行实测5 个 0.5s 的 sleep 任务多线程总耗时0.50s——5 个任务都在等等待时间全部重叠importtimefromthreadingimportThreaddefio_task(name,seconds0.5):time.sleep(seconds)# 模拟网络请求 / 磁盘 IO(阻塞时释放 GIL)returnf{name}donet0time.perf_counter()threads[Thread(targetio_task,args(ftask-{i},))foriinrange(5)]fortinthreads:t.start()fortinthreads:t.join()print(fIO 密集: 5 个 0.5s 的任务, 多线程总耗时{time.perf_counter()-t0:.2f}s)2.4 竞态条件与锁一个常见误解有 GIL 就不用操心线程安全。错。GIL 保护的是解释器内部状态不是你的业务逻辑。看一个取款函数检查余额和扣款是两步中间能被别的线程插队importtimefromthreadingimportThread,Lock balance100# 账户余额defunsafe_withdraw(amount10):globalbalanceifbalanceamount:# ① 检查: 还有钱, 能取time.sleep(0.001)# ② 窗口(模拟检查后还要做的其他工作)balance-amount# ③ 扣款returnTruereturnFalse厨房版库存清单上写着还有 10 个鸡蛋两个厨师同时看了一眼清单都认为够用各拿走 10 个——鸡蛋变负数。30 个线程并发取款余额只够 10 次成功实测无锁取款: 30 个线程并发各取 10 元, 余额只有 100 成功取款的线程数: 17 (余额只够 10 个) 最终余额: -70 - 17 个线程都通过了余额充足检查, 超额扣款17 个线程在各自检查时都看到余额充足然后全部扣款——余额打成 -70。修复用Lock锁把整个检查 扣款包成一个不可分割的整体术语叫原子操作——要么全做完要么没做别人插不进来balance2100lockLock()defsafe_withdraw(amount10):globalbalance2withlock:# 进门先拿钥匙, 出门自动还ifbalance2amount:time.sleep(0.001)balance2-amountreturnTruereturnFalse加锁取款: 同样 30 个线程 成功取款的线程数: 10 最终余额: 0 - 锁把检查扣款变成原子操作, 结果严格正确(代价: 并发退化成排队)两个工程要点锁的范围只锁balance - amount一行没用——检查在锁外竞态窗口还在。锁必须覆盖检查 修改的完整逻辑。用with lock:而不是手动acquire()/release()with 保证代码中途出异常时锁也一定释放不会把别人全锁死。2.5 asyncio单线程的并发asyncio 思路完全不同只有一个线程、一个事件循环一个总调度每个任务干到await等待点就主动交还控制权调度器趁等待的空档去推进别的任务。还是厨房只有一个厨师但他很聪明——炖上汤就转身去切菜水开了再回来下面。没有锁竞争没有线程切换开销单机轻松撑上万并发连接。完整实验代码含阻塞调用卡死事件循环的对照importasyncioimporttime# ---------- 1. 基本形态: async def 定义协程, await 让出控制权 ----------asyncdeffetch(name,seconds0.5):awaitasyncio.sleep(seconds)# 唯一的让出点: 在这里把控制权还给事件循环returnf{name}doneasyncdefdemo_gather():t0time.perf_counter()# 并发跑 5 个协程: 总耗时 ≈ 最长的那个, 而不是累加resultsawaitasyncio.gather(*(fetch(ftask-{i})foriinrange(5)))print(fasyncio.gather 并发 5 个 0.5s 任务:)print(f 总耗时{time.perf_counter()-t0:.2f}s (串行需要 2.5s))# ---------- 2. 协作式调度的坑: 阻塞调用会让整个循环停摆 ----------defcpu_heavy():普通的同步 CPU 函数: 没有任何 await, 跑起来事件循环就被卡死total0foriinrange(20_000_000):totalireturntotalasyncdefheart_beat(t0,n3):心跳: 每隔 0.1s 报一次事件循环的存活状态foriinrange(n):awaitasyncio.sleep(0.1)print(f [心跳{i1}] {time.perf_counter()-t0:.2f}s 循环还活着)asyncdefdemo_blocked():print(在协程里直接调用阻塞的 CPU 函数, 心跳会被完全卡住:)t0time.perf_counter()tasyncio.create_task(heart_beat(t0))cpu_heavy()# 阻塞 ~1-2s: 期间一次心跳都打不出来awaitt# 阻塞结束后, 3 次心跳挤在一起补发print( —— 看时间戳: 心跳全部挤在阻塞结束之后)# ---------- 3. 正确姿势: 阻塞任务丢进线程池 ----------asyncdefdemo_to_thread():print(用 asyncio.to_thread 把阻塞任务移出事件循环:)t0time.perf_counter()tasyncio.create_task(heart_beat(t0))awaitasyncio.to_thread(cpu_heavy)# Python 3.9: 事件循环继续转awaitt# 心跳全程每 0.1s 一次print( —— 看时间戳: 心跳均匀分布, 循环全程存活)if__name____main__:asyncio.run(demo_gather())asyncio.run(demo_blocked())asyncio.run(demo_to_thread())实测5 个 0.5s 任务总耗时0.51s——每个协程在await处让出等待时间全部重叠。但一个聪明厨师的方案有个致命契约他专心切菜的时候协程里出现同步阻塞调用整个厨房就停摆了——demo_blocked()里直接调用cpu_heavy()就是这种坑心跳任务明明每 0.1s 醒一次却全程一声不吭等阻塞结束才挤在一起补发。规避办法超时保护每个跨网络调用包一层wait_for到时间自动取消是 asyncio 服务的底线习惯。try:awaitasyncio.wait_for(fetch(slow,seconds2),timeout0.3)exceptasyncio.TimeoutError:print(自动取消慢任务)阻塞任务外抛Python 3.9await asyncio.to_thread(阻塞函数)把阻塞任务丢到线程池事件循环继续转。 
长河君 · 2026/9/6 6:07:56 · 阅读 1.2k Python 面向对象与并发模型
对象是组织代码的方式并发是组织执行的方式——这是 Python 最值得搞懂的两块。而且Python 并发绕不开一个东西GIL一、面向对象1.1 先搞懂类和对象是什么类是图纸对象是按图纸造出来的实物。一张账户图纸可以造出无数个账户每个账户有自己的主人数据。classAccount:def__init__(self,owner,balance0):self.ownerowner# 这个对象自己的属性self.balancebalance accAccount(小明,100)# 造出一个对象print(acc.owner)# 小明print(acc.balance)# 100三个初学必懂的点__init__是构造函数Account(小明, 100)执行时自动调用负责把新对象初始化好。self就是这个对象自己acc.deposit(50)其实等于Account.deposit(acc, 50)——Python 自动把acc填进第一个参数self。为什么用类把数据 操作这些数据的函数捆在一起账户的余额只该由账户的方法来改而不是散落在代码各处谁都能acc.balance -999。1.2 让对象能被 print__repr__直接 print 对象会得到__main__.Account object at 0x...调试时看不出内容。定义__repr__就能自定义打印的样子classAccount:def__init__(self,owner,balance0):self.ownerowner self._balancebalancedef__repr__(self):returnfAccount({self.owner!r},{self._balance})print(Account(小明,100))# Account(小明, 100)1.3 属性保护与 propertyPython 没有真正的 private私有变量靠约定单下划线_balance表示内部使用请别从外面直接改。但约定拦不住人真想要写入时校验用propertyclassAccount:def__init__(self,owner,balance0):self.ownerowner self._balancebalancepropertydefbalance(self):# 读: x acc.balance 走这里returnself._balancebalance.setterdefbalance(self,value):# 写: acc.balance x 走这里ifvalue0:raiseValueError(余额不能为负)self._balancevaluedefdeposit(self,amount):ifamount0:raiseValueError(存款必须为正)self._balanceamountreturnself._balance accAccount(小明,100)acc.balance-1# 触发 setter, 抛出: 余额不能为负顺带解释装饰器Python 里写在函数/方法头顶上、以开头的那行东西叫装饰器作用是给函数套一层壳再交还给你。property的壳做的事是把这个方法伪装成属性——外面用acc.balance读写不加括号实际却在偷偷走你写的方法。为什么值得这么做调用方始终写acc.balance将来要加校验只改 setter 一处所有调用代码一行不动——这就是 property 存在的理由。1.4 三种方法的分工classAccount:# ...前面的 __init__ / property / deposit 略...classmethoddefzero(cls,owner):替代构造器: 常用于从配置/数据库行创建对象returncls(owner,balance0)# cls 就是这个类自己staticmethoddefis_valid_amount(amount):纯工具函数, 跟实例/类状态都无关returnisinstance(amount,(int,float))andamount0Account.zero(新用户)# 不用先有实例, 直接用类调用Account.is_valid_amount(50)# True方法第一个参数用途普通方法self操作某个对象的状态绝大多数情况classmethodcls替代构造器、工厂方法staticmethod无纯工具函数1.5 继承、super、多态继承子类复用父类的全部代码再加自己的东西。classSavingsAccount(Account):# 括号里写父类, 就完成了继承def__init__(self,owner,balance0,rate0.02):super().__init__(owner,balance)# super() 调用父类的同名方法self.raterate# 子类新增的属性defadd_interest(self):# 子类新增的方法self._balance*(1self.rate)returnself._balancedefmonthly_statement(acc):多态: 只用 acc 的属性和方法, 不关心它具体是哪种账户returnf{acc.owner}:{acc.balance:.2f}sSavingsAccount(小明,1000,rate0.03)s.add_interest()print(monthly_statement(s))# 对 Account 的子类同样适用monthly_statement收的是Account传SavingsAccount进去照样工作——多态在 Python 里是免费的不用像 Java 那样先声明接口函数里只调了acc.owner和acc.balance任何有这两个属性的对象都能传进来。1.6 运算符重载让自定义对象像内置类型一样用1 2、len([1,2])、print(obj)这些写法背后其实都在调用对象身上的特殊方法名字都是双下划线包裹俗称魔法方法。自己实现了它们自己的对象就获得同样的能力classMoney:def__init__(self,amount,currency):self.amountamount self.currencycurrencydef__repr__(self):# print(m) 时的样子returnfMoney({self.amount}{self.currency})def__add__(self,other):# 让 Money 支持 号ifself.currency!other.currency:raiseValueError(币种不同不能直接相加)returnMoney(self.amountother.amount,self.currency)def__lt__(self,other):# 让 Money 支持 和 比较大小returnself.amountother.amountdef__bool__(self):# 定义 if m: 的行为returnself.amount!0a,bMoney(100,CNY),Money(58,CNY)print(ab)# Money(158 CNY)print(ab)# Trueprint(bool(Money(0,CNY)))# False同理实现__len__就能被len()量、实现__iter__就能被for遍历、实现__enter__/__exit__就能配合with使用。能干什么取决于实现了哪些方法。二、并发模型先直面 GIL2.1 先搞懂进程和线程是什么用一个厨房打比方程序 一份菜谱躺在硬盘上的代码进程 一间厨房程序跑起来后操作系统分给它独立的锅碗瓢盆——内存空间线程 厨房里的厨师每间厨房至少一个厨师同一厨房的厨师共享锅碗瓢盆CPU 核心 灶台的数量两个关键概念并行灶台不止一个多个厨师真的同时开火炒菜。并发灶台只有一个但厨师 A 炖上汤去等的时候厨师 B 插上来炒自己的菜——交替使用看起来像同时。2.2 GIL 是什么——用实验说话先补一句背景Python 代码不是直接在 CPU 上运行的而是先被翻译成字节码交给解释器执行。CPython最常用的官方解释器有个规定GILGlobal Interpreter Lock全局解释器锁同一时刻只允许一个线程执行 Python 字节码。用厨房的话说这间厨房不管有多少厨师灶台使用许可证只有一张——同一时刻只有一个厨师能真正开火其他人只能干等。空口无凭做实验。用同一个纯计算函数跑三遍串行 / 2 线程 / 2 进程importtimefrommultiprocessingimportPoolfromthreadingimportThread N80_000_000defcpu_bound(n):纯 CPU 计算: 一直占用灶台烧个不停的任务total0foriinrange(n):totali*ireturntotaldeftimeit(label,fn):t0time.perf_counter()fn()costtime.perf_counter()-t0print(f{label:12}{cost:.2f}s)if__name____main__:WORK[N,N]# 两个同样大小的任务# 1. 串行: 一个接一个做timeit(串行,lambda:[cpu_bound(n)forninWORK])# 2. 多线程: 同一进程内的两个厨师, 但许可证只有一张defrun_threads():ts[Thread(targetcpu_bound,args(n,))forninWORK]fortints:t.start()fortints:t.join()# join 等这个线程干完再往下走timeit(多线程,run_threads)# 3. 多进程: 再开一间厨房, 每间厨房有自己的许可证defrun_procs():withPool(2)asp:p.map(cpu_bound,WORK)timeit(多进程,run_procs)实测输出Python 3.13串行 10.51s 多线程 10.34s 多进程 6.29s三个结论多线程 10.34s ≈ 串行 10.51s几乎零加速。两张许可证不存在的——两个线程轮流持锁执行还多付了切换开销。多进程 6.29s——每个进程有独立的解释器和独立的 GIL两间厨房各有各的许可证真用上了 2 个核。没到理想的 5.25s 是因为开新厨房——进程创建和数据搬运——本身有开销。GIL 的准确表述它限制同一时刻执行 Python 字节码的线程数 1不是Python 不能多线程下面马上看到反例。2.3 GIL 什么时候会释放关键规则线程做 IO等网络、读写磁盘、time.sleep时会释放 GIL。回到厨房厨师把汤炖上、站在旁边等的时候等待不需要灶台许可证会让给别的厨师用。任务类型多线程有没有用原因IO 密集网络请求、读写文件✅ 有用等 IO 时释放 GIL其他线程干活CPU 密集数值计算、解析大 JSON❌ 没用字节码全程持锁退化成串行实测5 个 0.5s 的 sleep 任务多线程总耗时0.50s——5 个任务都在等等待时间全部重叠importtimefromthreadingimportThreaddefio_task(name,seconds0.5):time.sleep(seconds)# 模拟网络请求 / 磁盘 IO(阻塞时释放 GIL)returnf{name}donet0time.perf_counter()threads[Thread(targetio_task,args(ftask-{i},))foriinrange(5)]fortinthreads:t.start()fortinthreads:t.join()print(fIO 密集: 5 个 0.5s 的任务, 多线程总耗时{time.perf_counter()-t0:.2f}s)2.4 竞态条件与锁一个常见误解有 GIL 就不用操心线程安全。错。GIL 保护的是解释器内部状态不是你的业务逻辑。看一个取款函数检查余额和扣款是两步中间能被别的线程插队importtimefromthreadingimportThread,Lock balance100# 账户余额defunsafe_withdraw(amount10):globalbalanceifbalanceamount:# ① 检查: 还有钱, 能取time.sleep(0.001)# ② 窗口(模拟检查后还要做的其他工作)balance-amount# ③ 扣款returnTruereturnFalse厨房版库存清单上写着还有 10 个鸡蛋两个厨师同时看了一眼清单都认为够用各拿走 10 个——鸡蛋变负数。30 个线程并发取款余额只够 10 次成功实测无锁取款: 30 个线程并发各取 10 元, 余额只有 100 成功取款的线程数: 17 (余额只够 10 个) 最终余额: -70 - 17 个线程都通过了余额充足检查, 超额扣款17 个线程在各自检查时都看到余额充足然后全部扣款——余额打成 -70。修复用Lock锁把整个检查 扣款包成一个不可分割的整体术语叫原子操作——要么全做完要么没做别人插不进来balance2100lockLock()defsafe_withdraw(amount10):globalbalance2withlock:# 进门先拿钥匙, 出门自动还ifbalance2amount:time.sleep(0.001)balance2-amountreturnTruereturnFalse加锁取款: 同样 30 个线程 成功取款的线程数: 10 最终余额: 0 - 锁把检查扣款变成原子操作, 结果严格正确(代价: 并发退化成排队)两个工程要点锁的范围只锁balance - amount一行没用——检查在锁外竞态窗口还在。锁必须覆盖检查 修改的完整逻辑。用with lock:而不是手动acquire()/release()with 保证代码中途出异常时锁也一定释放不会把别人全锁死。2.5 asyncio单线程的并发asyncio 思路完全不同只有一个线程、一个事件循环一个总调度每个任务干到await等待点就主动交还控制权调度器趁等待的空档去推进别的任务。还是厨房只有一个厨师但他很聪明——炖上汤就转身去切菜水开了再回来下面。没有锁竞争没有线程切换开销单机轻松撑上万并发连接。完整实验代码含阻塞调用卡死事件循环的对照importasyncioimporttime# ---------- 1. 基本形态: async def 定义协程, await 让出控制权 ----------asyncdeffetch(name,seconds0.5):awaitasyncio.sleep(seconds)# 唯一的让出点: 在这里把控制权还给事件循环returnf{name}doneasyncdefdemo_gather():t0time.perf_counter()# 并发跑 5 个协程: 总耗时 ≈ 最长的那个, 而不是累加resultsawaitasyncio.gather(*(fetch(ftask-{i})foriinrange(5)))print(fasyncio.gather 并发 5 个 0.5s 任务:)print(f 总耗时{time.perf_counter()-t0:.2f}s (串行需要 2.5s))# ---------- 2. 协作式调度的坑: 阻塞调用会让整个循环停摆 ----------defcpu_heavy():普通的同步 CPU 函数: 没有任何 await, 跑起来事件循环就被卡死total0foriinrange(20_000_000):totalireturntotalasyncdefheart_beat(t0,n3):心跳: 每隔 0.1s 报一次事件循环的存活状态foriinrange(n):awaitasyncio.sleep(0.1)print(f [心跳{i1}] {time.perf_counter()-t0:.2f}s 循环还活着)asyncdefdemo_blocked():print(在协程里直接调用阻塞的 CPU 函数, 心跳会被完全卡住:)t0time.perf_counter()tasyncio.create_task(heart_beat(t0))cpu_heavy()# 阻塞 ~1-2s: 期间一次心跳都打不出来awaitt# 阻塞结束后, 3 次心跳挤在一起补发print( —— 看时间戳: 心跳全部挤在阻塞结束之后)# ---------- 3. 正确姿势: 阻塞任务丢进线程池 ----------asyncdefdemo_to_thread():print(用 asyncio.to_thread 把阻塞任务移出事件循环:)t0time.perf_counter()tasyncio.create_task(heart_beat(t0))awaitasyncio.to_thread(cpu_heavy)# Python 3.9: 事件循环继续转awaitt# 心跳全程每 0.1s 一次print( —— 看时间戳: 心跳均匀分布, 循环全程存活)if__name____main__:asyncio.run(demo_gather())asyncio.run(demo_blocked())asyncio.run(demo_to_thread())实测5 个 0.5s 任务总耗时0.51s——每个协程在await处让出等待时间全部重叠。但一个聪明厨师的方案有个致命契约他专心切菜的时候协程里出现同步阻塞调用整个厨房就停摆了——demo_blocked()里直接调用cpu_heavy()就是这种坑心跳任务明明每 0.1s 醒一次却全程一声不吭等阻塞结束才挤在一起补发。规避办法超时保护每个跨网络调用包一层wait_for到时间自动取消是 asyncio 服务的底线习惯。try:awaitasyncio.wait_for(fetch(slow,seconds2),timeout0.3)exceptasyncio.TimeoutError:print(自动取消慢任务)阻塞任务外抛Python 3.9await asyncio.to_thread(阻塞函数)把阻塞任务丢到线程池事件循环继续转。