第 51 课：并行计算：多进程（multiprocessing）

所属模块：模块十一：并发与并行难度：高级标签：parallelism, multiprocessing 上一课：线程安全与锁（Lock, RLock） 下一课：异步编程：asyncio 基础与 async/await

1. 学习目标

完成本课学习后，你将能够：

理解进程与线程的根本区别，以及何时应选择多进程进行并行计算。
使用 multiprocessing 模块创建、启动和管理多个进程。
掌握进程间通信（IPC）的基本方法，包括 Queue、Pipe 和共享内存。
了解全局解释器锁（GIL） 如何影响线程并行，以及多进程如何绕过此限制。
应用进程池（Pool） 简化大规模并行任务的执行。

2. 核心概念

进程 vs. 线程 想象一家工厂。一个进程就像一个完全独立的工厂，它有自己的地皮、厂房、工具和仓库（独立的内存空间）。而线程就像工厂里的工人，他们共享厂房和工具（共享内存），但各自执行不同的任务。因此，创建一个新进程（开新厂）的开销远大于创建一个新线程（雇一个新工人）。

为什么需要多进程？ Python 的 GIL（全局解释器锁）限制了同一进程中多个线程无法真正并行执行 CPU 密集型任务（如大量计算）。多进程通过创建多个独立的进程，每个进程拥有自己的 Python 解释器和内存空间，从而完全绕过了 GIL，可以利用多核 CPU 实现真正的并行计算。

multiprocessing 模块基础 这个模块的接口设计得与 threading 模块非常相似，使得从线程迁移到进程变得相对容易。主要类包括 Process（进程）、Pool（进程池）、Queue（队列）、Pipe（管道）等。

3. 代码示例

示例 1：基本多进程

import multiprocessing
import time
import os

def worker_function(name, duration):
    """一个简单的工作函数，模拟耗时任务"""
    print(f"进程 {name} (PID: {os.getpid()}) 开始工作...")
    time.sleep(duration)  # 模拟耗时计算
    print(f"进程 {name} (PID: {os.getpid()}) 工作完成。")
    return f"{name} 的结果"

if __name__ == '__main__':  # Windows/macOS 上必须使用此保护
    print(f"主进程 PID: {os.getpid()}")
    
    # 创建进程列表
    processes = []
    names = ["A", "B", "C"]
    durations = [2, 3, 1]

    for name, duration in zip(names, durations):
        # 创建一个进程对象，目标函数和参数
        p = multiprocessing.Process(target=worker_function, args=(name, duration))
        processes.append(p)
        p.start()  # 启动进程

    # 等待所有进程完成
    for p in processes:
        p.join()  # join() 会阻塞主进程，直到子进程p完成
        print(f"进程 {p.pid} 已退出。")

    print("所有进程已完成。")

解释：主进程创建了三个子进程。每个子进程独立运行 worker_function。注意 if __name__ == '__main__': 这个保护块，在 Windows 和 macOS 上这是必须的，它能防止子进程在导入模块时重复执行主代码。

示例 2：使用进程间通信（IPC）—— `Queue` 和 `Pipe`

import multiprocessing

def producer(queue):
    """生产者：将数据放入队列"""
    for i in range(5):
        item = f"数据项_{i}"
        print(f"生产者: 生产了 {item}")
        queue.put(item)  # 将数据放入队列
    queue.put(None)  # 发送结束信号

def consumer(queue):
    """消费者：从队列中取出数据"""
    while True:
        item = queue.get()  # 从队列获取数据，阻塞直到有数据
        if item is None:  # 检测到结束信号
            print("消费者: 收到结束信号，退出。")
            break
        print(f"消费者: 消费了 {item}")

if __name__ == '__main__':
    # 使用 Queue 进行通信
    q = multiprocessing.Queue()
    
    proc_producer = multiprocessing.Process(target=producer, args=(q,))
    proc_consumer = multiprocessing.Process(target=consumer, args=(q,))
    
    proc_producer.start()
    proc_consumer.start()
    
    proc_producer.join()
    proc_consumer.join()

    print("\n--- 使用 Pipe 通信 ---\n")

    # 使用 Pipe 进行通信 (一对一)
    parent_conn, child_conn = multiprocessing.Pipe()
    
    def sender(conn, data):
        conn.send(data)  # 通过连接发送数据
        conn.close()
    
    def receiver(conn):
        data = conn.recv()  # 通过连接接收数据
        print(f"接收方收到: {data}")
        conn.close()
    
    p_sender = multiprocessing.Process(target=sender, args=(parent_conn, "Hello from parent!"))
    p_receiver = multiprocessing.Process(target=receiver, args=(child_conn,))
    
    p_sender.start()
    p_receiver.start()
    p_sender.join()
    p_receiver.join()

解释：Queue 是一个线程和进程安全的 FIFO（先进先出）队列，非常适合生产者-消费者模型。Pipe() 返回两个连接对象，用于两个进程之间的双向通信，数据通过 send() 和 recv() 传递。

示例 3：使用共享内存

import multiprocessing

def update_shared_value(shared_value, lock, worker_id):
    """使用锁安全地更新共享值"""
    for _ in range(3):
        with lock:  # 获取锁，确保同一时间只有一个进程能修改
            shared_value.value += 1
            print(f"工作进程 {worker_id}: 值更新为 {shared_value.value}")

if __name__ == '__main__':
    # 创建一个共享的整数值
    shared_int = multiprocessing.Value('i', 0)  # 'i' 表示有符号整数
    lock = multiprocessing.Lock()
    
    processes = []
    for i in range(3):
        p = multiprocessing.Process(target=update_shared_value, args=(shared_int, lock, i))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()
    
    print(f"最终共享值: {shared_int.value}")
    # 预期输出：最终共享值应为 9 (3个进程 * 3次更新)

解释：multiprocessing.Value 和 multiprocessing.Array 提供了进程间共享的原始数据类型。由于多个进程可能同时尝试修改共享数据，必须使用锁（Lock） 来保证操作的原子性和数据一致性。

4. 实践练习

练习 1（基础）：并行计算列表平方和

创建一个程序，使用多进程计算一个大列表中所有数字的平方和。

主进程生成一个包含 1 到 10000 的列表。
将列表平均分成 4 个部分（假设有 4 个 CPU 核心）。
为每个部分创建一个子进程，计算该部分的平方和。
主进程收集所有子进程的结果，并计算总和。 预期输出：总和应为 10000 * 10001 * 20001 // 6（平方和公式）。

练习 2（进阶）：使用进程池进行任务映射

使用 multiprocessing.Pool 和 pool.map() 方法，重写练习 1。

定义一个计算单个数字平方的函数 square(n)。
使用进程池将此函数应用到列表的每个元素上。
对结果列表求和。要求：体会 pool.map() 的便捷性。

练习 3（挑战）：使用共享内存实现并发计数器

创建一个并发计数器，多个进程同时尝试将其递增 10000 次。

使用 multiprocessing.Value 创建一个初始值为 0 的共享整数。
创建多个进程，每个进程使用锁安全地对这个共享整数递增 10000 次。
最终验证共享整数的值是否正确。 预期输出：最终值应为 (进程数 * 10000)。

5. 常见错误

缺少 if __name__ == '__main__': 保护块：在 Windows 和 macOS 上，如果不加这个保护，当子进程启动时，它会重新导入主模块，导致无限创建子进程，程序崩溃。
误解共享数据：进程默认不共享内存。尝试像操作线程一样直接修改一个全局变量（例如，在子进程中修改列表或字典）是无效的，因为每个进程看到的是该变量的独立副本。必须使用 Value、Array 或通过 IPC 机制显式共享数据。
忘记关闭/加入进程：不调用 p.join() 或不关闭资源（如 Queue）可能导致僵尸进程或资源泄漏。
在 Windows 上使用 lambda 作为 target：在 Windows 上，使用 lambda 作为 Process 的目标函数有时会因为序列化问题而失败。尽量使用顶层定义的函数。
过度创建进程：创建和销毁进程的开销比线程大得多。对于大量小任务，使用进程池（Pool）来管理工作者进程池是更好的做法。

6. 小结

并行计算的核心工具：Python 的 multiprocessing 模块通过创建独立进程来绕过 GIL，是实现 CPU 密集型任务并行化的首选。
基本流程：创建 Process 对象 -> 调用 start() 启动 -> 使用 join() 等待完成。
进程间通信是必须的：由于内存隔离，你需要通过 Queue（多对多）、Pipe（一对一）或共享内存（Value/Array）来在进程间交换数据。
进程池简化任务管理：multiprocessing.Pool 和它的 map()、apply_async() 等方法可以高效地管理工作者进程，特别适合"分而治之"的任务。
安全第一：当使用共享内存时，锁（Lock） 是防止数据竞争的必需品。

理解多进程编程将为你打开充分利用现代计算机多核处理器能力的大门。下一课，我们将进入异步编程的世界，学习如何用单个线程高效处理大量 I/O 操作。

练习编辑器

python