当前位置：首页 > 文章列表 > 文章 > python教程 > Python多线程与多进程对比解析

Python多线程与多进程对比解析

2025-11-14 18:36:56 0浏览收藏

想提升Python并发性能？本文深入对比Python多线程与多进程在CPU密集型和IO密集型任务中的表现。受GIL（全局解释器锁）限制，多线程在CPU密集型任务中无法充分利用多核优势，性能甚至不如单线程；而多进程则能有效利用多核，显著提升计算速度。但在IO密集型任务中，多线程表现出色，线程在等待IO时释放GIL，允许其他线程执行，从而实现高并发。文章通过实际代码示例和测试数据，详细分析了两种并发方式的优缺点，并给出针对不同任务类型的选择建议，助你写出更高效的Python并发程序。

多进程在CPU密集型任务中性能优于多线程，因GIL限制多线程并行；而多线程在IO密集型任务中表现良好，适合高并发等待场景。

Python多线程性能测试对比 Python多线程与多进程效率分析

在Python中，多线程和多进程是实现并发编程的两种常见方式。但由于GIL（全局解释器锁）的存在，多线程在CPU密集型任务中表现不佳，而多进程则能真正利用多核优势。下面通过实际测试对比两者的性能差异。

测试场景设计

为了公平比较，我们设定两个典型任务：

CPU密集型任务：计算大量数字的平方和
IO密集型任务：模拟网络请求延迟（使用time.sleep）

分别用单线程、多线程、多进程执行，记录耗时。

CPU密集型任务性能对比

代码示例：

import threading
import multiprocessing
import time
<p>def cpu_task(n):
return sum(i * i for i in range(n))</p><p>def single_thread<em>cpu(n, loops):
for </em> in range(loops):
cpu_task(n)</p><p>def multi_thread<em>cpu(n, loops, threads=4):
def worker():
for </em> in range(loops // threads):
cpu_task(n)
threads<em>list = [threading.Thread(target=worker) for </em> in range(threads)]
for t in threads_list:
t.start()
for t in threads_list:
t.join()</p><p>def multi_process_cpu(n, loops, processes=4):
with multiprocessing.Pool(processes) as pool:
pool.map(cpu_task, [n] * loops)</p><h1>测试参数</h1><p>n = 10000
loops = 20</p><h1>单线程</h1><p>start = time.time()
single_thread_cpu(n, loops)
print(f"单线程耗时: {time.time() - start:.2f}s")</p><h1>多线程</h1><p>start = time.time()
multi_thread_cpu(n, loops)
print(f"多线程耗时: {time.time() - start:.2f}s")</p><h1>多进程</h1><p>start = time.time()
multi_process_cpu(n, loops)
print(f"多进程耗时: {time.time() - start:.2f}s")</p>

结果分析：

多线程耗时接近甚至超过单线程，因为GIL限制了并行执行
多进程显著快于前两者，充分利用多核CPU

IO密集型任务性能对比

模拟IO操作（如网络请求）：

import time
import threading
import multiprocessing
<p>def io_task(seconds):
time.sleep(seconds)</p><p>def single_thread<em>io(loops, sec=0.1):
for </em> in range(loops):
io_task(sec)</p><p>def multi_thread<em>io(loops, sec=0.1, threads=4):
def worker():
for </em> in range(loops // threads):
io_task(sec)
threads<em>list = [threading.Thread(target=worker) for </em> in range(threads)]
for t in threads_list:
t.start()
for t in threads_list:
t.join()</p><p>def multi_process_io(loops, sec=0.1, processes=4):
with multiprocessing.Pool(processes) as pool:
pool.map(io_task, [sec] * loops)</p><h1>测试参数</h1><p>loops = 40
sec = 0.1</p><h1>单线程</h1><p>start = time.time()
single_thread_io(loops, sec)
print(f"IO-单线程耗时: {time.time() - start:.2f}s")</p><h1>多线程</h1><p>start = time.time()
multi_thread_io(loops, sec)
print(f"IO-多线程耗时: {time.time() - start:.2f}s")</p><h1>多进程</h1><p>start = time.time()
multi_process_io(loops, sec)
print(f"IO-多进程耗时: {time.time() - start:.2f}s")</p>

结果分析：