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

Python 并发编程：asyncio vs threading vs multiprocessing

news 2026/7/22 10:07:42

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 实验结果分析

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