3. Python多进程实战：multiprocessing模块、进程间通信（Queue/Pipe）、进程池Pool的使用

聊到Python并发，很多人第一反应就是多线程。但说实话，GIL这把锁让多线程在CPU密集型任务上基本成了摆设。我早年做爬虫调度系统时就踩过这个坑——开了16个线程，CPU利用率死活上不去，一看监控，8个核都在排队等GIL释放。

后来我换了个思路：既然线程被GIL卡脖子，那干脆上多进程。每个进程有独立的Python解释器和内存空间，GIL互不干扰。今天我们就来聊聊Python的multiprocessing模块，怎么用它做真正的并行计算。

3.1 进程的创建与基础操作

multiprocessing.Process的用法和threading.Thread几乎一样。你只需要定义一个函数，然后把它丢进进程里跑就行。

from multiprocessing import Process
import os

def worker(name):
    print(f"进程 {name} 启动，PID: {os.getpid()}")
    # 模拟计算任务
    result = sum(i * i for i in range(10_000_000))
    print(f"进程 {name} 完成，结果: {result}")

if __name__ == "__main__":
    processes = []
    for i in range(4):
        p = Process(target=worker, args=(f"Worker-{i}",))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    print("所有进程执行完毕")

这里有个细节我提醒一下：if __name__ == "__main__": 这行不是摆设。Windows下创建子进程时会重新导入模块，如果没有这个保护，子进程会递归创建进程，直接炸掉。我在给团队做培训时，至少有一半人第一次跑多进程代码都栽在这上面。

3.2 进程间通信：Queue vs Pipe

多进程最大的痛点就是数据共享。进程之间内存不互通，你不能像线程那样直接读写全局变量。那怎么办？两种官方方案：Queue 和 Pipe。

3.2.1 Queue：生产者-消费者模式的首选

multiprocessing.Queue 是线程安全且进程安全的队列。底层用了管道加锁，支持多个生产者和多个消费者。我个人习惯在大多数场景下优先用Queue，因为它用起来最省心。

from multiprocessing import Process, Queue
import time

def producer(q):
    for i in range(5):
        item = f"数据-{i}"
        q.put(item)
        print(f"生产: {item}")
        time.sleep(0.5)

def consumer(q):
    while True:
        item = q.get()
        if item is None:  # 哨兵值，表示结束
            break
        print(f"消费: {item}")
        time.sleep(1)

if __name__ == "__main__":
    q = Queue()
    p1 = Process(target=producer, args=(q,))
    p2 = Process(target=consumer, args=(q,))

    p1.start()
    p2.start()

    p1.join()
    q.put(None)  # 发送结束信号
    p2.join()

3.2.2 Pipe：双向通信，但要注意数据完整性

Pipe本质上是一对读写端，适合两个进程之间的点对点通信。它比Queue轻量，但没有内置的锁机制，多个进程同时写Pipe会导致数据错乱。

from multiprocessing import Process, Pipe

def send_msg(conn):
    conn.send("Hello from child")
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p = Process(target=send_msg, args=(child_conn,))
    p.start()
    msg = parent_conn.recv()
    print(f"收到消息: {msg}")
    p.join()

嗯，这里要注意：Pipe的send()和recv()是阻塞的。如果两端同时发送或接收，可能会死锁。我建议只在两个进程间用Pipe，而且最好明确谁发谁收，别搞成双向同时读写。

3.3 进程池Pool：管理海量进程的利器

你想想看，如果任务有几百个，难道要手动创建几百个Process对象吗？显然不现实。进程池就是干这个的——它维护一组工作进程，你把任务丢进去，它自动调度执行。

from multiprocessing import Pool
import time

def cpu_intensive(n):
    """模拟CPU密集型计算"""
    return sum(i * i for i in range(n))

if __name__ == "__main__":
    # 创建进程池，默认使用CPU核心数
    with Pool(processes=4) as pool:
        # map：同步方式，阻塞直到所有任务完成
        results = pool.map(cpu_intensive, [10_000_000] * 8)
        print(f"同步结果: {results}")

        # apply_async：异步方式，立即返回
        async_results = [pool.apply_async(cpu_intensive, args=(10_000_000,)) 
                        for _ in range(8)]
        async_results = [r.get() for r in async_results]
        print(f"异步结果: {async_results}")

3.3.1 Pool的三种提交方式

• apply()：同步执行，阻塞等待结果。适合单个任务。
• apply_async()：异步执行，立即返回AsyncResult对象。适合批量提交。
• map() / map_async()：类似内置map，将可迭代对象分发给进程池。适合数据并行。
• starmap() / starmap_async()：类似map，但支持多参数元组。

我个人最常用的是apply_async配合回调函数。为什么呢？因为你可以一边提交任务，一边处理已完成的结果，不用等所有任务跑完。

def callback(result):
    print(f"任务完成，结果: {result}")

with Pool(4) as pool:
    for i in range(10):
        pool.apply_async(cpu_intensive, args=(i * 1_000_000,), callback=callback)
    pool.close()
    pool.join()

3.4 进程间同步：Lock与共享内存

虽然多进程不共享内存，但有些场景下你还是需要同步——比如多个进程同时写同一个文件，或者更新共享计数器。这时候就需要multiprocessing.Lock和Value/Array。

from multiprocessing import Process, Lock, Value

def increment(counter, lock):
    for _ in range(1000):
        with lock:
            counter.value += 1

if __name__ == "__main__":
    counter = Value('i', 0)  # 'i' 表示有符号整数
    lock = Lock()
    processes = [Process(target=increment, args=(counter, lock)) 
                for _ in range(10)]

    for p in processes:
        p.start()
    for p in processes:
        p.join()

    print(f"最终计数: {counter.value}")  # 应该输出 10000

3.5 知识体系总览

下面这张图把本章的核心知识点串起来了。你可以把它当作多进程开发的路线图——从进程创建，到通信，到池化管理，再到同步控制。

3.6 实战建议与避坑总结

最后，我把这些年用多进程踩过的坑和总结的经验列出来，希望能帮你少走弯路。

• 进程数不是越多越好：CPU密集型任务，进程数超过核心数反而会降低性能。I/O密集型可以适当多开，但别超过核心数的2倍。
• Queue的容量要监控：如果生产者速度远大于消费者，Queue会无限膨胀，最终撑爆内存。我习惯在Queue上设置maxsize，或者用qsize()做限流。
• 避免频繁创建销毁进程：进程创建开销很大（fork或spawn）。能用Pool就别手动创建Process，复用才是王道。
• 子进程异常要处理：子进程崩溃不会自动通知父进程。建议在子进程里加try/except，把异常通过Queue传回父进程。
• Windows下注意spawn模式：Linux默认fork，Windows默认spawn。spawn模式下子进程会重新导入模块，全局变量和资源需要重新初始化。