Python并发设计原则_复杂度控制说明【指导】

Python并发设计核心是控制复杂度而非堆砌工具:按IO特征选模型(asyncio适配IO密集、multiprocessing用于CPU密集),避免混用模型,优先用通信代替共享状态,显式处理错误,强制超时与限流。

Python并发设计的核心不是堆砌工具,而是用最轻量、最贴合场景的方式解决问题。过度使用线程、协程或进程,反而会放大调试难度、资源争用和逻辑耦合——控制复杂度,比实现并发本身更重要。

按IO特征选模型,不为“高级”而升级

多数Web服务、API调用、文件读写属于IO密集型,asyncio + aiohttp/aiofiles足够高效;CPU密集任务(如数值计算、图像处理)才需考虑multiprocessing。混用threadingasyncio(比如在协程里调用阻塞IO)会破坏事件循环,引入隐性同步点,让性能和可读性双降。

  • HTTP请求多且慢?优先用aiohttp.ClientSession,别用requests配线程池
  • 要并行处理100个本地JSON文件?concurrent.futures.ProcessPoolExecutor可能不如单线程+json.load快——先测再选
  • 有遗留同步库(如某些数据库驱动)必须调用?用loop.run_in_executor隔离,但明确标注“此处退化为同步”

共享状态能免则免,通信优于共享

全局变量、类属性、模块级dict在并发下极易引发竞态。Python的GIL不保护用户数据结构,list.appenddict[key] = value都不是原子操作。

  • 协程间传递数据?用asyncio.Queue,它线程安全且适配异步生命周期
  • 多进程需要共享配置?序列化后通过multiprocessing.Manager.dict()或启动参数注入,而非动态修改全局dict
  • 真需计数/开关状态?用asyncio.Lockmultiprocessing.Value,并把临界区缩到最小——例如只锁赋值,不锁整个计算过程

错误处理必须显式覆盖,不依赖“自动传播”

协程中未捕获异常会静默消失;线程崩溃默认不中断主线程;子进程退出码被忽略就等于失败不可见。并发放大了错误隐蔽性。

  • 所有asyncio.create_task()都应搭配asyncio.gather(..., return_exceptions=True)或单独await并try-catch
  • concurrent.futures.Executor.submit()时,务必调用future.result()(它会重新抛出执行中的异常)
  • 子进程启动后,检查proc.returncode,非零时记录stderr输出——不要只靠proc.wait()返回True/False

监控与限流是并发系统的标配,不是优化项

没有超时、没有并发数限制、没有成功率统计的并发,等于裸奔。复杂度常来自失控的资源膨胀,而非代码本身。

  • 所有网络调用加timeout:aiohttp用timeout=ClientTimeout(total=5),requests用timeout=(3, 7)
  • 限制最大并发数:asyncio.Semaphore(10)包住关键协程,ThreadPoolExecutor(max_workers=8)设合理上限
  • 关键路径埋点:记录每批任务的平均耗时、失败率、队列堆积深度——用time.perf_counter()time.time()更准

不复杂但容易忽略。