编辑“︁Python 进程池”︁

{{DISPLAYTITLE:Python进程池}}

'''Python进程池'''是[[Python并发编程]]中用于管理和复用多个进程的高效工具，通过[[multiprocessing]]模块的<code>Pool</code>类实现。它允许开发者将任务分配给一组预创建的进程，避免频繁创建/销毁进程的开销，特别适合处理[[CPU密集型任务]]。

== 核心概念 ==
进程池的核心机制是“任务队列+工作进程”模型：
* '''主进程'''将任务放入队列
* '''工作进程'''从队列获取任务并执行
* '''结果'''通过队列或共享内存返回

<mermaid>
graph LR
    A[主进程] -->|提交任务| B[任务队列]
    B --> C[进程1]
    B --> D[进程2]
    B --> E[...]
    C --> F[结果队列]
    D --> F
    E --> F
    F --> A
</mermaid>

== 基本用法 ==
=== 创建进程池 ===
使用<code>multiprocessing.Pool</code>初始化：

<syntaxhighlight lang="python">
from multiprocessing import Pool

# 创建包含4个工作进程的池
pool = Pool(processes=4)
</syntaxhighlight>

=== 任务提交方法 ===
{| class="wikitable"
|+ 常用方法对比
! 方法 !! 描述 !! 返回类型
|-
| <code>apply()</code> || 同步阻塞式提交 || 直接返回结果
|-
| <code>apply_async()</code> || 异步非阻塞提交 || <code>AsyncResult</code>对象
|-
| <code>map()</code> || 批量同步处理 || 结果列表
|-
| <code>map_async()</code> || 批量异步处理 || <code>AsyncResult</code>对象
|}

=== 示例：计算平方 ===
同步方式：
<syntaxhighlight lang="python">
def square(x):
    return x * x

if __name__ == '__main__':
    with Pool(4) as p:
        results = p.map(square, range(10))
    print(results)  # 输出: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
</syntaxhighlight>

异步方式：
<syntaxhighlight lang="python">
if __name__ == '__main__':
    with Pool(4) as p:
        async_result = p.map_async(square, range(10))
        print("继续执行其他操作...")
        results = async_result.get()  # 阻塞获取结果
    print(results)
</syntaxhighlight>

== 高级特性 ==
=== 回调函数 ===
可为异步任务添加完成回调：
<syntaxhighlight lang="python">
def callback(result):
    print(f"任务完成，结果为: {result}")

if __name__ == '__main__':
    with Pool(2) as p:
        p.apply_async(square, (5,), callback=callback)
        p.close()
        p.join()
</syntaxhighlight>

=== 错误处理 ===
通过<code>error_callback</code>捕获异常：
<syntaxhighlight lang="python">
def error_handler(e):
    print(f"发生错误: {str(e)}")

def risky_operation(x):
    if x == 3:
        raise ValueError("特殊值错误")
    return x * x

if __name__ == '__main__':
    with Pool(2) as p:
        p.apply_async(risky_operation, (3,), 
                     error_callback=error_handler)
        p.close()
        p.join()
</syntaxhighlight>

== 性能优化 ==
=== 进程数选择 ===
最优进程数通常为CPU核心数，可通过以下公式估算：
<math>
N_{optimal} = N_{cores} \times (1 + \frac{W}{C})
</math>
其中：
* <math>W</math>：等待时间
* <math>C</math>：计算时间

=== 数据分块 ===
处理大数据集时合理设置<code>chunksize</code>：
<syntaxhighlight lang="python">
# 将10000个任务分成每块100个
pool.map(func, range(10000), chunksize=100)
</syntaxhighlight>

== 实际应用案例 ==
=== 图像批量处理 ===
<syntaxhighlight lang="python">
from PIL import Image
import os

def process_image(filename):
    with Image.open(filename) as img:
        img = img.filter(ImageFilter.GaussianBlur(2))
        img.save(f"processed_{filename}")

if __name__ == '__main__':
    image_files = ['a.jpg', 'b.jpg', 'c.jpg']
    with Pool() as p:
        p.map(process_image, image_files)
</syntaxhighlight>

=== 科学计算加速 ===
并行计算<math>\pi</math>的蒙特卡洛近似：
<syntaxhighlight lang="python">
import random

def monte_carlo(n):
    inside = 0
    for _ in range(n):
        x, y = random.random(), random.random()
        if x**2 + y**2 <= 1:
            inside += 1
    return 4 * inside / n

if __name__ == '__main__':
    with Pool(4) as p:
        results = p.map(monte_carlo, [10**6]*4)
    pi_approx = sum(results)/len(results)
    print(f"π ≈ {pi_approx}")
</syntaxhighlight>

== 常见问题 ==
=== 进程间通信 ===
* 使用<code>Queue</code>、<code>Pipe</code>或共享内存
* 避免传递不可pickle的对象

=== Windows系统限制 ===
在Windows下必须保护入口代码：
<syntaxhighlight lang="python">
if __name__ == '__main__':
    # 进程池代码
</syntaxhighlight>

== 最佳实践 ==
# 始终使用<code>with</code>语句自动管理资源
# 避免在子进程中修改全局状态
# 对于IO密集型任务考虑[[线程池]]替代
# 监控进程状态：<code>pool._processes</code>、<code>pool._taskqueue</code>

== 参见 ==
* [[Python多进程编程]]
* [[Python全局解释器锁(GIL)]]
* [[Python并行计算模式]]

[[Category:Python并发编程]]
[[Category:Python标准库]]

[[Category:编程语言]]
[[Category:Python]]
[[Category:Python 并发编程]]