Python并发编程实战：多线程vs多进程性能对比，一篇文章让你彻底选对方案

引言

在Python开发中，当我们需要同时处理多个任务时，常常会面临一个经典的选择：使用多线程还是多进程？由于全局解释器锁（GIL）的存在，Python的多线程并不能真正利用多核CPU实现并行计算，这使得很多开发者对多线程望而却步。但多线程真的毫无用武之地吗？多进程又是否在所有场景下都更优？

本文将通过真实的CPU密集型与IO密集型任务，编写可运行的代码，对比单线程、多线程与多进程的执行效率。你会看到，在不同的任务类型下，两者的表现截然不同。最终，我们将总结出清晰的选择策略，帮助你在实际项目中做出正确决策。

1. 核心概念：GIL、线程与进程

1.1 Python的全局解释器锁（GIL）

GIL是CPython解释器中的一个互斥锁，它确保同一时刻只有一个线程在执行Python字节码。这意味着，即使在多核CPU上，多线程也无法并行执行Python代码，它只能通过快速切换线程来模拟并发。因此，对于纯Python的CPU密集型任务，多线程几乎无法提升性能，甚至可能因为线程切换的额外开销而变得更慢。

1.2 多线程适用场景

虽然GIL限制了CPU密集型任务的并行性，但在IO密集型场景（如网络请求、文件读写、数据库查询等）中，线程在执行IO操作时会释放GIL，让其他线程有机会执行。此时多线程可以实现真正的并发，极大减少等待时间，提升程序吞吐量。

1.3 多进程适用场景

多进程通过创建独立的解释器进程来绕过GIL，每个进程拥有自己的内存空间和GIL锁，因此可以真正利用多核CPU进行并行计算。对于CPU密集型任务，多进程能显著缩短执行时间。但进程的创建和销毁开销较大，进程间通信也比线程间共享内存复杂得多。

2. 实战对比：CPU密集型与IO密集型任务

下面我们设计两个典型任务，分别测试单线程、多线程（ThreadPoolExecutor）和多进程（ProcessPoolExecutor）的执行时间。所有代码均可直接复制运行。

2.1 实验环境准备

import time import math from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed from functools import partial

2.2 任务定义：CPU密集型

用一个计算密集的函数模拟：计算一个大数的阶乘（使用math.factorial，纯Python循环也可，这里用标准库更高效），重复执行多次。

def cpu_bound_task(n): """CPU密集型任务：计算 n 的阶乘""" return math.factorial(n)

为了简单起见，我们让每个任务计算一个较大的数，例如 100000，并重复执行若干次，记录总耗时。

2.3 任务定义：IO密集型

模拟多个网络请求，使用time.sleep来代表IO等待。实际项目中的网络IO、磁盘IO等，原理相同。

def io_bound_task(seconds): """IO密集型任务：模拟网络或磁盘等待""" time.sleep(seconds) return seconds

2.4 性能测试工具函数

编写一个通用的测试函数，接收任务函数、参数列表和执行器类型，返回总耗时。

def run_test(task_func, args_list, executor_type=None, max_workers=4): """ 执行测试 :param task_func: 任务函数 :param args_list: 参数列表，每个元素是一个元组，解包后传给任务函数 :param executor_type: 'thread' 或 'process'，None 表示单线程顺序执行 :param max_workers: 最大工作线程/进程数 :return: 总耗时(秒) """ start = time.time() if executor_type is None: # 单线程顺序执行 results = [task_func(*args) for args in args_list] else: Executor = ThreadPoolExecutor if executor_type == 'thread' else ProcessPoolExecutor with Executor(max_workers=max_workers) as executor: futures = {executor.submit(task_func, *args): args for args in args_list} # 等待所有任务完成，这里简单遍历 results = [future.result() for future in as_completed(futures)] end = time.time() return end - start

注意：ProcessPoolExecutor在提交任务时，函数和参数必须可序列化（pickle），我们使用的函数和参数都满足。

3. 完整测试代码与结果分析

整合上述代码，分别对CPU密集和IO密集任务进行测试。

if __name__ == "__main__": # 测试参数设置 cpu_n = 100000 # 计算阶乘的数 cpu_task_count = 12 # 任务数量（足够将CPU吃满） io_sleep = 0.1 # 每次IO等待时间（秒） io_task_count = 50 # IO任务数量，体现并发优势 workers = 8 # 最大线程/进程数（根据实际CPU核心数调整） # 准备参数列表 cpu_args = [(cpu_n,) for _ in range(cpu_task_count)] io_args = [(io_sleep,) for _ in range(io_task_count)] print("=" * 60) print("Python多线程 vs 多进程性能对比") print(f"CPU任务：计算 {cpu_n} 的阶乘，共 {cpu_task_count} 次") print(f"IO任务： 每次 sleep {io_sleep} 秒，共 {io_task_count} 次") print(f"并发 workers 数：{workers}") print("=" * 60) # CPU密集型测试 print("\n--- CPU密集型任务 ---") t_cpu_seq = run_test(cpu_bound_task, cpu_args, executor_type=None) print(f"单线程耗时： {t_cpu_seq:.2f} 秒") t_cpu_thread = run_test(cpu_bound_task, cpu_args, executor_type='thread', max_workers=workers) print(f"多线程耗时： {t_cpu_thread:.2f} 秒") t_cpu_process = run_test(cpu_bound_task, cpu_args, executor_type='process', max_workers=workers) print(f"多进程耗时： {t_cpu_process:.2f} 秒") # IO密集型测试 print("\n--- IO密集型任务 ---") t_io_seq = run_test(io_bound_task, io_args, executor_type=None) print(f"单线程耗时： {t_io_seq:.2f} 秒") t_io_thread = run_test(io_bound_task, io_args, executor_type='thread', max_workers=workers) print(f"多线程耗时： {t_io_thread:.2f} 秒") t_io_process = run_test(io_bound_task, io_args, executor_type='process', max_workers=workers) print(f"多进程耗时： {t_io_process:.2f} 秒")

运行结果示例（在你的机器上可能略有不同，但趋势一致）：

```