当前位置： 首页 > news >正文

Python 并发编程：asyncio vs threading vs multiprocessing

news 2026/4/15 23:01:37

Python 并发编程：asyncio vs threading vs multiprocessing

核心结论

  • asyncio：适合 I/O 密集型任务，内存开销小，单线程并发
  • threading：适合 I/O 密集型任务，多线程并发，需注意 GIL 限制
  • multiprocessing：适合 CPU 密集型任务，多进程并发，无 GIL 限制
  • 性能对比：I/O 密集型任务 asyncio > threading > multiprocessing；CPU 密集型任务 multiprocessing > threading ≈ asyncio

一、并发编程基础

1.1 并发与并行的区别

  • 并发：多个任务交替执行，宏观上同时进行
  • 并行：多个任务同时执行，微观上同时进行
  • Python 中的并发模型：
    • 线程（threading）：多线程并发，受 GIL 限制
    • 进程（multiprocessing）：多进程并行，无 GIL 限制
    • 协程（asyncio）：单线程并发，基于事件循环

1.2 GIL（全局解释器锁）的影响

  • GIL：Python 解释器的全局锁，同一时刻只能有一个线程执行 Python 字节码
  • 影响：
    • 多线程在 CPU 密集型任务中无法真正并行
    • I/O 密集型任务中，线程会释放 GIL，因此仍有并发优势
    • 多进程不受 GIL 影响，可实现真正并行

二、asyncio 详解

2.1 基本概念

  • 协程：可暂停执行的函数，通过async def定义
  • 事件循环：管理协程的执行，处理 I/O 操作
  • Future：表示异步操作的结果
  • Task：Future 的子类，用于执行协程

2.2 代码示例

import asyncio import time async def fetch_data(url, delay): """模拟网络请求""" print(f"开始获取 {url} 的数据") await asyncio.sleep(delay) print(f"完成获取 {url} 的数据") return f"{url} 的数据" async def main(): """主协程""" start_time = time.time() # 并发执行多个协程 tasks = [ fetch_data("https://api.example.com/data1", 2), fetch_data("https://api.example.com/data2", 3), fetch_data("https://api.example.com/data3", 1) ] results = await asyncio.gather(*tasks) print(f"所有请求完成，结果: {results}") print(f"总耗时: {time.time() - start_time:.2f} 秒") if __name__ == "__main__": asyncio.run(main())

2.3 性能分析

  • 优点：
    • 内存开销小，无需线程/进程切换
    • 适合高并发 I/O 操作
    • 编程模型清晰，避免回调地狱
  • 缺点：
    • 需使用异步库，不能直接调用同步函数
    • 不适合 CPU 密集型任务
    • 学习曲线较陡峭

三、threading 详解

3.1 基本概念

  • 线程：轻量级进程，共享内存空间
  • Thread 类：创建和管理线程
  • Lock：线程同步原语，防止资源竞争
  • ThreadPoolExecutor：线程池，管理线程生命周期

3.2 代码示例

import threading import time from concurrent.futures import ThreadPoolExecutor def fetch_data(url, delay): """模拟网络请求""" print(f"开始获取 {url} 的数据") time.sleep(delay) print(f"完成获取 {url} 的数据") return f"{url} 的数据" def main(): """主线程""" start_time = time.time() # 使用线程池 with ThreadPoolExecutor(max_workers=3) as executor: futures = [ executor.submit(fetch_data, "https://api.example.com/data1", 2), executor.submit(fetch_data, "https://api.example.com/data2", 3), executor.submit(fetch_data, "https://api.example.com/data3", 1) ] results = [future.result() for future in futures] print(f"所有请求完成，结果: {results}") print(f"总耗时: {time.time() - start_time:.2f} 秒") if __name__ == "__main__": main()

3.3 性能分析

  • 优点：
    • 适合 I/O 密集型任务
    • 编程模型简单，易于理解
    • 可直接调用同步函数
  • 缺点：
    • 受 GIL 限制，CPU 密集型任务性能受限
    • 线程切换开销较大
    • 需注意线程安全问题

四、multiprocessing 详解

4.1 基本概念

  • 进程：独立的执行环境，有自己的内存空间
  • Process 类：创建和管理进程
  • Queue：进程间通信机制
  • Pool：进程池，管理进程生命周期

4.2 代码示例

import multiprocessing import time from concurrent.futures import ProcessPoolExecutor def compute_intensive_task(n): """模拟 CPU 密集型任务""" print(f"开始计算任务 {n}") result = 0 for i in range(10**7): result += i print(f"完成计算任务 {n}") return result def main(): """主进程""" start_time = time.time() # 使用进程池 with ProcessPoolExecutor(max_workers=3) as executor: futures = [ executor.submit(compute_intensive_task, 1), executor.submit(compute_intensive_task, 2), executor.submit(compute_intensive_task, 3) ] results = [future.result() for future in futures] print(f"所有计算完成，结果: {results}") print(f"总耗时: {time.time() - start_time:.2f} 秒") if __name__ == "__main__": main()

4.3 性能分析

  • 优点：
    • 无 GIL 限制，适合 CPU 密集型任务
    • 真正的并行执行
    • 进程间相互独立，安全性高
  • 缺点：
    • 内存开销大，每个进程有独立内存空间
    • 进程间通信开销较大
    • 启动和管理开销较大

五、性能对比实验

5.1 I/O 密集型任务对比

import asyncio import threading import multiprocessing import time from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor # 模拟 I/O 密集型任务 def io_task(delay): time.sleep(delay) return delay async def async_io_task(delay): await asyncio.sleep(delay) return delay # 测试 I/O 密集型任务 def test_io_performance(): tasks = [1] * 10 # 10个任务，每个任务延迟1秒 # 同步执行 start = time.time() for task in tasks: io_task(task) sync_time = time.time() - start print(f"同步执行时间: {sync_time:.2f} 秒") # asyncio 执行 async def async_main(): start = time.time() await asyncio.gather(*[async_io_task(task) for task in tasks]) return time.time() - start async_time = asyncio.run(async_main()) print(f"asyncio 执行时间: {async_time:.2f} 秒") # threading 执行 start = time.time() with ThreadPoolExecutor(max_workers=10) as executor: executor.map(io_task, tasks) thread_time = time.time() - start print(f"threading 执行时间: {thread_time:.2f} 秒") # multiprocessing 执行 start = time.time() with ProcessPoolExecutor(max_workers=10) as executor: executor.map(io_task, tasks) process_time = time.time() - start print(f"multiprocessing 执行时间: {process_time:.2f} 秒") if __name__ == "__main__": test_io_performance()

5.2 CPU 密集型任务对比

import threading import multiprocessing import time from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor # 模拟 CPU 密集型任务 def cpu_task(n): result = 0 for i in range(n): result += i return result # 测试 CPU 密集型任务 def test_cpu_performance(): tasks = [10**7] * 4 # 4个任务，每个任务计算10^7次 # 同步执行 start = time.time() for task in tasks: cpu_task(task) sync_time = time.time() - start print(f"同步执行时间: {sync_time:.2f} 秒") # threading 执行 start = time.time() with ThreadPoolExecutor(max_workers=4) as executor: executor.map(cpu_task, tasks) thread_time = time.time() - start print(f"threading 执行时间: {thread_time:.2f} 秒") # multiprocessing 执行 start = time.time() with ProcessPoolExecutor(max_workers=4) as executor: executor.map(cpu_task, tasks) process_time = time.time() - start print(f"multiprocessing 执行时间: {process_time:.2f} 秒") if __name__ == "__main__": test_cpu_performance()

5.3 实验结果分析

任务类型同步执行asynciothreadingmultiprocessing
I/O 密集型（10个任务，每个1秒）10.0+秒~1.0秒~1.0秒~1.0秒+
CPU 密集型（4个任务，每个10^7次计算）4.0+秒~4.0秒~4.0秒~1.0秒

结论：

  • I/O 密集型任务：asyncio 和 threading 性能接近，multiprocessing 略慢（进程启动开销）
  • CPU 密集型任务：multiprocessing 性能显著优于 threading 和 asyncio（无 GIL 限制）

六、最佳实践建议

6.1 选择合适的并发模型

  • asyncio：
    • 适合：网络请求、文件 I/O、数据库操作等 I/O 密集型任务
    • 场景：Web 服务器、爬虫、API 调用
  • threading：
    • 适合：I/O 密集型任务，特别是需要调用同步库的场景
    • 场景：传统 I/O 操作、GUI 应用
  • multiprocessing：
    • 适合：CPU 密集型任务，如数据处理、模型训练
    • 场景：科学计算、图像处理、机器学习

6.2 性能优化技巧

  • asyncio：
    • 使用异步库（aiohttp、aiomysql 等）
    • 避免在协程中执行阻塞操作
    • 合理设置事件循环
  • threading：
    • 使用线程池管理线程
    • 减少线程间通信和同步操作
    • 注意 GIL 影响
  • multiprocessing：
    • 使用进程池管理进程
    • 减少进程间通信开销
    • 合理设置进程数（通常为 CPU 核心数）

6.3 代码质量保证

  • 错误处理：在并发代码中妥善处理异常
  • 资源管理：确保所有资源正确释放
  • 测试：编写并发测试用例，验证正确性和性能
  • 监控：监控并发任务的执行状态和资源使用

七、总结

Python 提供了三种主要的并发编程模型：asyncio、threading 和 multiprocessing。每种模型都有其适用场景和优缺点：

  • asyncio：单线程协程，适合 I/O 密集型任务，内存开销小，性能优异
  • threading：多线程并发，适合 I/O 密集型任务，编程模型简单，但受 GIL 限制
  • multiprocessing：多进程并行，适合 CPU 密集型任务，无 GIL 限制，但内存开销大

在实际应用中，应根据任务类型、性能要求和代码复杂度选择合适的并发模型。对于复杂系统，也可以结合使用多种并发模型，充分发挥各自的优势。

技术演进的内在逻辑：Python 的并发模型从 threading 到 multiprocessing，再到 asyncio，反映了对性能和编程体验的不断追求。每种模型都解决了特定场景下的问题，共同构成了 Python 强大的并发编程生态。

http://www.jsqmd.com/news/647063/

相关文章：

  • MATLAB柱状图进阶：5分钟搞定分组数据+数值标注（附完整代码）
  • 安装阿帕奇maven的相关配置
  • 生成式AI应用用户流失率飙升的真正原因：不是模型不准，而是这6个隐性体验缺口未被填补
  • 即插即用系列 | CVPR 2024 FADC：频域自适应采样，从根源消除分割“棋盘格”
  • LeRobot实战指南：3步构建你的机器人学习工作流
  • 人大金仓 KingbaseES V8 数据库 Docker 部署指南
  • 从零到一的无人机DIY手记（一）：配件采购与兼容性排雷
  • 别再混淆了！一文搞懂IP协议号47（GRE）、6（TCP）、17（UDP）的区别与联系
  • CSS如何快速实现网站换肤功能_利用CSS变量重置全局颜色方案
  • 保姆级教程：用Python和PyTorch Geometric从零搭建GCN，实战DEAP情感脑电识别
  • Unity游戏资源逆向解析：从APK到Asset的完整提取指南
  • 多模态旅游推荐到底难在哪？SITS2026团队亲述：97.3%的失败源于这4类跨模态对齐陷阱
  • 【工业控制系统网络安全系列课程】第2课-工业控制系统的网络安全风险-过程控制漏洞利用（二）典型漏洞利用路径-物理过程影响攻击
  • 【ETestDEV5教程37】测试开发之代码搜索
  • 专科大二学生的变成学习规划和愿景
  • 从键盘敲击到游戏手柄：libusb中断传输（Interrupt Transfer）在HID设备开发中的实战指南
  • LTspice新手必看：从零搭建12V转5V降压整流电路的完整仿真指南
  • 为什么92%的多模态POC在长尾测试集上失败？：基于LLaVA-1.6/InternVL 2.5的17万条长尾case归因分析与增量蒸馏修复框架
  • OBS Studio实战：SRT推流配置全解析与性能优化
  • Umi-CUT：三分钟掌握批量图片去黑边的终极解决方案
  • 2025届必备的五大AI辅助写作神器解析与推荐
  • GD32F450时钟配置避坑指南：从8MHz晶振到200MHz主频的完整流程（含代码详解）
  • BilibiliDown：3步完成B站视频下载的完整免费解决方案
  • ABB机器人通讯实战——四元数与欧拉角互转的编程实现
  • 我用了一周 Hermes Agent，整理出这十件必做的事
  • 测试数据管理模型服务化
  • 7.8%复合增速！无人机管理软件未来六年发展路径清晰
  • 实时AI视频生成已突破24fps？2026奇点大会现场Demo实测：端侧部署方案、WebGPU加速路径与iOS/Android兼容性避坑指南
  • 以数字化服务为核心，爱毕业aibiye等机构持续优化用户体验，赢得广泛认可
  • Archery权限管理实战：从RD到DBA的多级审批流程详解（附避坑指南）