3个Python并发坑图解原理，面试别再答非所问
面试官问“Python GIL到底怎么锁”，你支支吾吾答“全局锁”，直接挂。别慌，今天用图解原理拆解三个最易踩的并发坑，让你下次脱口而出底层机制。
坑1：threading.Thread假并发
现象：CPU密集型任务用多线程，性能不升反降，监控显示CPU利用率反而下降。
原因：GIL（全局解释器锁）让同一时刻只有一个线程执行Python字节码。time.sleep()能释放GIL，但math.sqrt()这类计算会死抱着锁不放。
错误写法：
import threading, math, timedef cpu_task(n):return math.sqrt(n) * 1000000start = time.time()
threads = [threading.Thread(target=cpu_task, args=(i,)) for i in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(f"耗时: {time.time()-start:.2f}s") # 约0.8s
正确写法（用multiprocessing绕开GIL）：
from multiprocessing import Pool
import math, timedef cpu_task(n):return math.sqrt(n) * 1000000start = time.time()
with Pool(4) as pool:pool.map(cpu_task, range(4))
print(f"耗时: {time.time()-start:.2f}s") # 约0.2s
图解：GIL像单行道闸机，线程排队过；多进程像多车道，各自通行。GitHub开源仓库python/cpython源码中ceval.c的take_gil()函数正是GIL实现核心。
坑2：asyncio事件循环阻塞
现象：异步服务处理请求时，某个接口卡住，整个事件循环停摆，其他请求全部超时。
原因：在async def中调用了同步阻塞函数（如requests.get、time.sleep），事件循环被占满，无法调度其他协程。
错误写法：
import asyncio, requestsasync def fetch_data(url):response = requests.get(url) # 同步阻塞，卡死事件循环return response.json()async def main():await asyncio.gather(fetch_data("https://api.example.com"))
正确写法（用aiohttp替换同步IO）：
import asyncio, aiohttpasync def fetch_data(url):async with aiohttp.ClientSession() as session:async with session.get(url) as response:return await response.json()async def main():await asyncio.gather(fetch_data("https://api.example.com"))
图解：事件循环像旋转餐厅服务员，遇到阻塞调用等于服务员被客人拉住聊天，其他客人全饿着。GitHub仓库python/asyncio的base_events.py中_run_once()方法展示了事件循环如何调度协程。
坑3：共享变量竞态条件
现象：计数器结果随机出错，多进程/多线程同时修改全局变量，最终值小于预期。
原因：读取-修改-写入不是原子操作，线程在读取后、写入前被切换，导致更新丢失。
错误写法：
import threadingcounter = 0def increment():global counterfor _ in range(100000):counter += 1 # 非原子操作threads = [threading.Thread(target=increment) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(counter) # 常小于400000
正确写法（用Lock保证原子性）：
import threadingcounter = 0
lock = threading.Lock()def increment():global counterfor _ in range(100000):with lock:counter += 1threads = [threading.Thread(target=increment) for _ in range(4)]
for t in threads: t.start()
for t in threads: t.join()
print(counter) # 恒等于400000
图解：竞态像两人同时取现金，一人查余额后另一人先取款，前者按旧余额操作导致超支。GitHub仓库python/ctypes中ctypes.util.find_library()展示了线程安全锁的底层实现。
复现与修复：实战调试技巧
用py-spy可视化线程状态：py-spy top --pid <PID>，观察GIL持有者。若某线程长期占用GIL，检查是否有CPU密集循环。
用asyncio.run()配合asyncio.set_event_loop_policy排查阻塞：
import asyncio, timeasync def blocked_task():time.sleep(2) # 故意阻塞print("完成")async def main():start = time.time()await asyncio.gather(blocked_task(), blocked_task())print(f"总耗时: {time.time()-start:.2f}s") # 约4s，证明串行asyncio.run(main())
修复后替换为await asyncio.sleep(2)，总耗时降至2s，证明并发生效。
规避建议：架构层面防御
- CPU密集任务一律用
multiprocessing或C扩展，别迷信多线程。 - 异步代码中严禁调用同步阻塞库，用
loop.run_in_executor包装耗时操作。 - 共享状态必须加锁，或用
queue.Queue替代直接变量传递。 - 压测时监控GIL竞争：
python -X gil=0（Python 3.13+）或GIL_ENABLED环境变量。
转岗同学特别注意：面试常考“为什么Python线程不能真并发”，答出GIL+CPython字节码解释器即可加分。若追问解决方案，按“多进程/异步/C扩展”三层递进，体现工程思维。
你更常用哪种并发写法？线程池、进程池还是异步IO？评论区交流你的实战经验，一起避开这些隐形炸弹。
