企业官网建设流程全解析

4 个线程没跑满 4 个核：GIL 限制了什么，什么时候该换多进程？——GIL 到底是干嘛的？多线程为什么在 CPU 密集场景反而更慢？Python 实战应用开发

先看一个非常典型的现场：一台 4 核 CPU 的机器，代码写了threading开了 4 个线程去并行计算，结果任务管理器里 CPU 占用率只有 100% 左右，4 个核心没有一个被真正跑满，反而像在排队。如果你也遇到过这个问题，不用怀疑电脑坏了，也不用怀疑threading写错了，问题出在 Python 解释器的一个底层机制：GIL。

GIL 这个名词在 Python 面试里出现频率极高，但很多人对它的理解停留在“Python 多线程是假的并发”这个结论上。这不算错，但不够准确。GIL 并不是把多线程完全废掉，它限制的是 CPU 密集场景下的并行能力；而在 IO 密集场景里，多线程依然非常有用。这篇文章会从 GIL 的工作原理讲起，然后用可运行的实验代码对比多线程和多进程在 CPU 密集、IO 密集两类任务上的表现，最后给出工程上的选型建议和排查方法。

1. GIL 核心概念速览

GIL 不是 Python 语言本身的限制，而是 CPython 解释器的实现细节。它保证解释器内部的引用计数等内存管理操作不会被多线程同时打断，从而避免复杂的锁竞争。代价就是：在单个进程内，多个线程无法真正同时执行计算密集型代码。

从工程角度看，关键不是“GIL 存在”，而是“GIL 在什么场景下影响性能”。理解这一点，才能回答一个经典问题：Python 多线程到底什么时候用？

2. 先看现象：多线程为什么跑不满 CPU

先准备一个简单的验证脚本。这个脚本用 4 个线程分别执行一个纯计算任务，统计总耗时。

import threading import time def cpu_task(n): result = 0 for i in range(n): result += i * i return result def run_threads(): n = 5_000_000 threads = [] start = time.perf_counter() for _ in range(4): t = threading.Thread(target=cpu_task, args=(n,)) threads.append(t) t.start() for t in threads: t.join() elapsed = time.perf_counter() - start print(f"多线程 CPU 密集任务耗时: {elapsed:.4f} 秒") if __name__ == "__main__": run_threads()

运行这段代码时注意观察任务管理器或系统监视器。你会看到整机 CPU 占用率远低于预期。单线程跑这个任务需要约 1 秒的话，4 个线程同时跑并不会变成 0.25 秒，而是接近 4 秒。

原因是 GIL 强制线程轮流执行。每个线程获得 GIL 后执行一段时间，然后切换到下一个线程。切换本身也有开销，所以多线程在 CPU 密集场景下不但不能加速，反而会比单线程更慢。

接下来看多进程版本：

import multiprocessing as mp import time def cpu_task(n): result = 0 for i in range(n): result += i * i return result def run_processes(): n = 5_000_000 processes = [] start = time.perf_counter() for _ in range(4): p = mp.Process(target=cpu_task, args=(n,)) processes.append(p) p.start() for p in processes: p.join() elapsed = time.perf_counter() - start print(f"多进程 CPU 密集任务耗时: {elapsed:.4f} 秒") if __name__ == "__main__": run_processes()

每个进程有独立的 Python 解释器和独立的内存空间，也各自持有一把 GIL，因此 4 个进程可以真正并行跑在 4 个核心上。同一个任务，多进程版本通常会接近单线程耗时的 1/4，比如单线程 4 秒，多进程 1 秒多一点。

这个对比就是 GIL 最直观的体现。

3. 实验环境准备

本文的代码不依赖任何第三方库，使用 Python 标准库即可完成。

建议环境：

验证 Python 版本：

python --version

如果是 Linux / macOS，可能需要用python3 --version。

另外，建议在项目目录下新建一个concurrency_benchmark文件夹，把实验脚本按类型分开存放：

concurrency_benchmark/ ├── gil_thread_cpu.py ├── gil_process_cpu.py ├── gil_thread_io.py ├── gil_process_pool.py └── async_demo.py

这样后续跑性能对比时不容易把脚本搞混。

4. 核心实验：GIL 到底限制了什么

4.1 实验一：CPU 密集任务，单线程 vs 多线程 vs 多进程

完整的对比脚本如下：

import threading import multiprocessing as mp import time def cpu_task(n): result = 0 for i in range(n): result += i * i return result def single_thread(n): start = time.perf_counter() cpu_task(n) cpu_task(n) cpu_task(n) cpu_task(n) print(f"单线程顺序执行: {time.perf_counter() - start:.4f} 秒") def multi_thread(n): threads = [] start = time.perf_counter() for _ in range(4): t = threading.Thread(target=cpu_task, args=(n,)) threads.append(t) t.start() for t in threads: t.join() print(f"多线程执行: {time.perf_counter() - start:.4f} 秒") def multi_process(n): processes = [] start = time.perf_counter() for _ in range(4): p = mp.Process(target=cpu_task, args=(n,)) processes.append(p) p.start() for p in processes: p.join() print(f"多进程执行: {time.perf_counter() - start:.4f} 秒") if __name__ == "__main__": N = 5_000_000 single_thread(N) multi_thread(N) multi_process(N)

运行结果通常类似于：

单线程顺序执行: 2.4416 秒 多线程执行: 2.6931 秒 多进程执行: 0.7294 秒

说明：

• 多线程比单线程还慢一点，这是 GIL 线程切换的开销。
• 多进程明显更快，能接近线性加速。

如果测试机器是 2 核 CPU，多进程加速比可能只有 1.7 倍左右，这是正常现象。如果是 8 核机器，开 8 个进程效果更明显。

4.2 实验二：IO 密集任务，多线程能加速吗

IO 密集任务的特点是：大部分时间都花在等待外部资源上，比如网络响应、文件读写、数据库查询。这些等待期间不占用 CPU，GIL 会被释放，因此多线程可以产生实际的并发效果。

用time.sleep模拟 IO 等待：

import threading import time def io_task(seconds): time.sleep(seconds) def single_thread_io(seconds): start = time.perf_counter() for _ in range(10): io_task(seconds) print(f"单线程 IO 任务: {time.perf_counter() - start:.4f} 秒") def multi_thread_io(seconds): threads = [] start = time.perf_counter() for _ in range(10): t = threading.Thread(target=io_task, args=(seconds,)) threads.append(t) t.start() for t in threads: t.join() print(f"多线程 IO 任务: {time.perf_counter() - start:.4f} 秒") if __name__ == "__main__": single_thread_io(0.5) multi_thread_io(0.5)

单线程版本需要大约 5 秒，多线程版本只需要约 0.5 秒。这个提升来自并发等待，而不是并行计算。

真实场景中，网络爬虫、批量发送 HTTP 请求、批量读取小文件，都是典型 IO 密集任务，用concurrent.futures.ThreadPoolExecutor可以轻松提升效率。

5. 实战工具：concurrent.futures 统一管理线程与进程

实际工程中很少手动管理threading.Thread和multiprocessing.Process，更推荐使用concurrent.futures提供了统一的线程池和进程池接口。

5.1 线程池处理 IO 密集任务

以批量请求 HTTP 接口为例：

import time import urllib.request from concurrent.futures import ThreadPoolExecutor def fetch_url(url): with urllib.request.urlopen(url, timeout=10) as response: return response.status def main(): urls = [ "https://www.baidu.com", "https://www.zhihu.com", "https://www.bilibili.com", ] * 20 start = time.perf_counter() with ThreadPoolExecutor(max_workers=10) as executor: results = list(executor.map(fetch_url, urls)) print(f"状态码数量: {len(results)}, 耗时: {time.perf_counter() - start:.4f} 秒") if __name__ == "__main__": main()

关键点：

• ThreadPoolExecutor(max_workers=10)控制并发数，不是越大越好，过大会造成连接数过多。
• executor.map会按输入顺序返回结果，方便批量处理。

5.2 进程池处理 CPU 密集任务

用进程池改写第 4.1 节的 CPU 密集测试：

import time from concurrent.futures import ProcessPoolExecutor def cpu_task(n): result = 0 for i in range(n): result += i * i return result def main(): n = 5_000_000 start = time.perf_counter() with ProcessPoolExecutor(max_workers=4) as executor: futures = [executor.submit(cpu_task, n) for _ in range(4)] results = [f.result() for f in futures] print(f"进程池执行耗时: {time.perf_counter() - start:.4f} 秒") print(f"任务结果数量: {len(results)}") if __name__ == "__main__": main()

ProcessPoolExecutor会自动创建多进程，任务分发和结果收集比直接multiprocessing更简洁。

使用进程池要注意一个事项：任务函数和参数必须可以被 pickle 序列化。如果需要在进程间共享更复杂的数据，需要借助multiprocessing.Manager、Queue或共享内存。

6. 多进程进阶：进程间通信与共享数据

多进程不共享内存，因此进程之间交换数据需要额外的通信机制。常见的方案有三种。

6.1 multiprocessing.Queue 队列

import multiprocessing as mp def worker(q): q.put("任务完成") if __name__ == "__main__": q = mp.Queue() p = mp.Process(target=worker, args=(q,)) p.start() print(q.get()) p.join()

Queue适合生产者消费者模型，多个任务进程把结果写入队列，主进程统一消费。

6.2 multiprocessing.Pipe 管道

import multiprocessing as mp def worker(conn): conn.send("data from child") conn.close() if __name__ == "__main__": parent_conn, child_conn = mp.Pipe() p = mp.Process(target=worker, args=(child_conn,)) p.start() print(parent_conn.recv()) p.join()

Pipe适合两个进程之间的双向通信，性能比Queue高一些。

6.3 multiprocessing.Manager 共享对象

import multiprocessing as mp def worker(namespace, key): namespace.data[key] = "processed" if __name__ == "__main__": manager = mp.Manager() namespace = manager.Namespace() namespace.data = manager.dict() p = mp.Process(target=worker, args=(namespace, "key1")) p.start() p.join() print(namespace.data)

Manager可以创建供多进程共享的字典、列表、命名空间等对象，使用方便，但性能较低。适合数据量不大、操作不频繁的场景。

7. asyncio：比多线程更轻的并发方案

在 IO 密集场景中，asyncio是比多线程更轻量的选择。它基于事件循环和协程，不使用线程，因此完全没有线程上下文切换的消耗，也不受 GIL 的排队切换干扰。

一个简单的 asyncio 并发示例：

import asyncio import time async def io_task(seconds): await asyncio.sleep(seconds) async def main(): start = time.perf_counter() await asyncio.gather( io_task(0.5), io_task(0.5), io_task(0.5), io_task(0.5), ) print(f"asyncio 并发耗时: {time.perf_counter() - start:.4f} 秒") if __name__ == "__main__": asyncio.run(main())

io_task内的await asyncio.sleep()会主动让出控制权，事件循环继续执行其他协程。整个过程只需要一个线程，内存占用远小于开几十个线程。

asyncio 适合以下场景：

• 大量网络请求，比如爬虫、API 调用。
• WebSocket 长连接服务。
• 高并发网关或消息推送服务。

但它有一个学习门槛：所有 IO 操作都必须使用异步版本，不能直接调用同步的requests.get()或time.sleep()。如果代码里出现了同步阻塞调用，整个事件循环都会被卡住。

8. 资源占用与性能观察方法

8.1 如何确认 GIL 是否影响程序

判断一段代码是否受 GIL 限制，核心是看任务类型：

• CPU 计算密集：受 GIL 影响明显。
• IO 等待密集：不受 GIL 影响，多线程有效。
• 混合型任务：如果 CPU 计算部分占比高，整体收益会打折。

也可以用cProfile观察程序热点：

python -m cProfile -s cumulative gil_thread_cpu.py

如果热点集中在纯计算函数上，并且多线程版本比单线程版更慢，基本可以断定是 GIL 导致。

8.2 CPU 占用率与核数关系

在 4 核机器上：

观察方法：Windows 任务管理器打开“性能”页，Linux 用htop，macOS 用“活动监视器”。

8.3 降低性能开销的常见手段

• 将嵌套 for 循环中的核心计算改为 NumPy，NumPy 底层是 C 语言，释放 GIL，可以获得接近多核的性能。
• 使用 PyPy 解释器，它对 GIL 的实现和调度方式与 CPython 不同，部分计算密集场景有惊喜。
• Python 3.13 的实验性自由线程构建（free-threaded build），可以先做小规模验证，但生产环境还不建议直接切。
• 把重计算部分写成 C 扩展或使用ctypes，在 C 层面释放 GIL。

9. 常见问题与排查方法

其中最容易踩坑的是 pickle 序列化问题。使用ProcessPoolExecutor和multiprocessing时，任务函数必须定义在模块顶层，且参数必须可序列化。闭包、局部函数、Lambda 表达式在进程池中都无法正常工作。

10. 实战场景选型总结

10.1 用多线程的场景

• 网络爬虫，大量并发请求等待响应。
• 大批量文件读写，每文件耗时短但数量多。
• 数据库批量查询，单次查询有几十毫秒延迟。
• 调用第三方 API 或 SDK。

这类任务的共同点是：代码本身没有重度计算，等待时间是瓶颈，多线程可以用少量线程管理大量阻塞操作。

10.2 用多进程的场景

• 视频转码、图像批量处理、音频分析。
• 大数据量排序、搜索、特征提取。
• 多租户任务隔离，避免单任务崩溃影响整个服务。

多进程的代价是内存开销更大，进程间通信更慢。如果任务数据量巨大，还要考虑数据传输和序列化成本。

10.3 用 asyncio 的场景

• 高并发长连接服务。
• WebSocket 聊天服务。
• 微服务网关。
• 大量轻量级 IO 操作，线程数容易过多时。

asyncio 特别适合单个进程内管理上万个并发连接的需求，这种规模用线程池是不现实的。

10.4 混合方案

实际工程中经常是组合使用：

• 用ProcessPoolExecutor做 CPU 密集计算。
• 用ThreadPoolExecutor或 asyncio 做网络等待。
• 使用multiprocessing.Queue在计算进程和 IO 线程之间传递数据。

示例：批量下载图片并做压缩处理。

import time import urllib.request from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor def download_image(url): data = urllib.request.urlopen(url, timeout=10).read() return data def process_image(data): # 这里可以写入压缩或格式转换逻辑 result = len(data) return result def main(): urls = [ "https://www.python.org/static/img/python-logo.png", ] * 20 with ThreadPoolExecutor(max_workers=8) as thread_pool: image_data = list(thread_pool.map(download_image, urls)) with ProcessPoolExecutor(max_workers=4) as process_pool: results = list(process_pool.map(process_image, image_data)) print(f"处理完成 {len(results)} 张图片") if __name__ == "__main__": main()

这个例子展示了 IO 密集部分用线程池并发下载、CPU 密集部分用进程池并行处理的典型正交组合。

11. 最佳实践与工程建议

11.1 先写小尺寸基准测试

不要一上来就开几十个线程跑全量数据。先构造一个最小规模的 benchmark，跑通单线程、多线程、多进程三组对比，确认瓶颈到底是 CPU 还是 IO。

推荐记录三组数据：

单线程耗时: 多线程耗时: 多进程耗时:

当多线程耗时明显小于单线程时，说明任务属于 IO 密集；当多线程耗时接近或超过单线程时，任务属于 CPU 密集，应改用多进程。

11.2 多进程要合理设置进程数

进程数不是越多越好。常见经验值是CPU 核数或CPU 核数 + 1。如果任务本身还有其他 OS 进程在竞争资源，盲目开满核数不一定最快。

获取 CPU 核数：

import multiprocessing as mp import os print(f"CPU 逻辑核数: {os.cpu_count()}") print(f"CPU 物理核数: {mp.cpu_count()}")

11.3 保留一套最小可运行脚本

把第 4.1 节的对比脚本保存为benchmark_template.py，以后遇到不确定的场景，可以直接修改任务函数复用。这个模板会告诉你当前硬件和 Python 版本下，多线程、多进程、asyncio 到底谁更合适。

11.4 注意生产环境部署差异

Windows 下多进程启动使用spawn方式，子进程会重新导入主模块，因此if __name__ == "__main__"保护必须写。Linux 下默认fork，行为略有不同但同样建议带上主模块保护。

在容器环境运行时，os.cpu_count()返回的可能不是宿主机的真实核数，需要确认容器的 CPU 配额。

12. 写在最后

这个问题的本质是：GIL 存在是为了保护 CPython 的内存模型，它没有限制多线程在 IO 密集场景下的表现，只在 CPU 密集场景下让多线程无法并行利用多核。

下次遇到“4 个线程没跑满 4 个核”时，先不要急着给线程加数量，而是判断任务的类型。CPU 密集场景直接用多进程或 NumPy，IO 密集场景放心用多线程，需要海量并发连接时优先考虑 asyncio。把这套判断逻辑记下来，比死记“Python 多线程是假的”这个结论要实用得多。

建议把文章中的三个实验脚本都跑一遍，尤其是第 4.1 节的单线程多线程多进程对比，跑一次之后对 GIL 的理解会非常直观。