Likanug // lab
返回文章列表

文章

技术分享

Python中的线程池和进程池

关于线程池和进程池你需要知道的基础

#Python
目录
  1. Python 中的线程池（ThreadPoolExecutor）和进程池（ProcessPoolExecutor）都是 concurrent.futures 模块提供的高层抽象，用于并发执行任务。它们使用方式类似，但适用场景不同。
  2. ✅ 1. 使用方式对比
  3. 👉 线程池：ThreadPoolExecutor
  4. 👉 进程池：ProcessPoolExecutor
  5. ✅ 2. 使用场景对比
  6. ✅ 3. 实际应用场景
  7. 🔸 线程池（适合 IO 密集型任务）：
  8. 🔸 进程池（适合 CPU 密集型任务）：
  9. ✅ 4. GIL（全局解释器锁）解释
  10. ✅ 5. 附加提示
  11. 在 IO 密集型应用 中，线程的瓶颈通常不在 CPU，而在等待 I/O（如网络、磁盘、数据库）完成。因此线程池大小 可以远大于 CPU 核心数。
  12. 🧠 配置原则
  13. 1. 基础经验公式：
  14. 2. 建议配置（经验值）
  15. 🧪 实践建议
  16. ✅ 示例（测试并发网络）
  17. 📎 参考文章

Python 中的线程池（ThreadPoolExecutor）和进程池（ProcessPoolExecutor）都是 concurrent.futures 模块提供的高层抽象，用于并发执行任务。它们使用方式类似，但适用场景不同。#

✅ 1. 使用方式对比#

👉 线程池：ThreadPoolExecutor#

from concurrent.futures import ThreadPoolExecutor

def task(x):
    return x * x

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [executor.submit(task, i) for i in range(10)]
    results = [f.result() for f in futures]
    print(results)
  • submit(func, *args)：异步提交任务，返回 Future
  • map(func, iterable)：类似内建 map，自动收集结果

👉 进程池：ProcessPoolExecutor#

from concurrent.futures import ProcessPoolExecutor

def task(x):
    return x * x

with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(task, range(10)))
    print(results)
  • 接口与线程池几乎一样
  • 使用多进程（multiprocessing），避免 GIL 限制

✅ 2. 使用场景对比#

特性`ThreadPoolExecutor`（线程池）`ProcessPoolExecutor`（进程池）
并发模型多线程多进程
是否受 GIL 限制✅ 是❌ 否
启动开销小（共享内存）大（进程启动、拷贝内存）
适合任务类型IO 密集型（如网络、文件）CPU 密集型（如数学计算）
共享数据可共享全局变量（但需线程安全）不共享内存（使用 `multiprocessing.Queue/Pipe`）
错误传播通过 `Future.exception()`同样支持

✅ 3. 实际应用场景#

🔸 线程池（适合 IO 密集型任务）：#

  • 网络爬虫
  • 文件读取/写入
  • 网络请求并发处理（如批量 requests.get）
  • 数据库连接池

🔸 进程池（适合 CPU 密集型任务）：#

  • 图像/视频处理
  • 数据压缩、加密、哈希计算
  • 自然语言处理（大规模文本清洗、向量化）
  • 数据分析（Pandas 大量计算）

✅ 4. GIL（全局解释器锁）解释#

Python 的标准实现（CPython）中有 GIL，导致多线程不能真正实现并行执行 CPU 密集任务。因此：

  • IO 密集任务：线程池效果好
  • CPU 密集任务：推荐使用进程池

✅ 5. 附加提示#

  • 可用 as_completed() 或 wait() 管理多个 Future 结果
  • Python 3.11 引入了 ExceptionGroup 便于处理多个任务抛出异常
  • 高级库如 joblib, ray, dask 封装了更多高级并行计算能力

在 IO 密集型应用 中，线程的瓶颈通常不在 CPU，而在等待 I/O（如网络、磁盘、数据库）完成。因此线程池大小 可以远大于 CPU 核心数。#

🧠 配置原则#

1. 基础经验公式：#

线程数 ≈ CPU 核心数 × (1 + IO 等待时间 / CPU 计算时间) 对于典型的 I/O 密集型任务，这个比值通常是 >> 1，所以： 👉 线程数可配置为 CPU 核心数的 2~10 倍甚至更高

2. 建议配置（经验值）#

应用场景推荐线程池大小
纯 CPU 密集型核心数 或 核心数 + 1
网络 I/O 密集型（如爬虫）**CPU 核心数 × 5\~10 或更多**
高并发数据库访问CPU 核心数 × 2\~4（受数据库连接限制）

你可以先试试：

import os
cpu_count = os.cpu_count() or 4
thread_pool_size = cpu_count * 5  # IO密集型建议起点

🧪 实践建议#

  • 观察程序运行时的 CPU 占用和线程等待时间：使用 psutil、top、htop 等工具。
  • Python 中用 ThreadPoolExecutor 或 asyncio：对于大量 I/O 并发（如下载、测速），ThreadPoolExecutor(max_workers=50~200) 是常见配置。
  • 小心资源限制：网络连接上限、目标服务抗压能力，甚至文件句柄数（ulimit）。

✅ 示例（测试并发网络）#

from concurrent.futures import ThreadPoolExecutor, as_completed
import os

cpu_count = os.cpu_count() or 4
max_workers = cpu_count * 8  # IO密集型：每核8个线程

with ThreadPoolExecutor(max_workers=max_workers) as executor:
    futures = [executor.submit(your_network_task, node) for node in nodes]
    for f in as_completed(futures):
        process_result(f.result())

如你正在测试 10K 个节点的网络延迟，推荐：

  • 初始线程池大小 100~300，视你主机网络带宽与远程服务响应能力而定；
  • 动态分批分段测试，减少瞬时并发过大带来的问题。 如需，我可帮你设计一个高效分批测试方案。

📎 参考文章#