4、多进程编程：multiprocessing模块详解、进程间通信（Queue、Pipe）、进程池的使用

聊到多进程，我得先说说自己的一个心路历程。早些年我刚接触Python并发编程时，总觉得多线程是万能的。直到有一次线上服务扛不住，CPU利用率死活上不去，我才意识到——GIL这玩意儿，真不是闹着玩的。说白了，计算密集型的任务，多线程就是个摆设。这时候，多进程才是真正的解药。

4.1 为什么需要多进程？

Python的多线程因为GIL的存在，同一时刻只有一个线程能执行字节码。你想想看，如果你有8个CPU核心，多线程只能用一个，那剩下的7个不就浪费了吗？

多进程就不一样了。每个进程都有独立的Python解释器和内存空间，没有GIL的限制。每个进程都能跑满一个CPU核心。我曾在一次数据处理任务中，用多进程把处理时间从40分钟压缩到了6分钟——嗯，效果就是这么明显。

核心区别：

  • 多线程：适合I/O密集型任务（网络请求、文件读写）
  • 多进程：适合CPU密集型任务（计算、数据处理、图像渲染）

4.2 multiprocessing模块基础

Python的multiprocessing模块，接口设计跟threading非常像。如果你写过线程，上手进程几乎零成本。

from multiprocessing import Process
import os

def worker(name):
    print(f"子进程 {name} 启动，PID: {os.getpid()}")

if __name__ == '__main__':
    print(f"主进程 PID: {os.getpid()}")
    p = Process(target=worker, args=('worker-1',))
    p.start()
    p.join()
    print("子进程结束")

这里有个坑，我一开始就踩过——if __name__ == '__main__'这行不能省。Windows下创建子进程会重新导入模块，没有这个保护，子进程会无限递归创建进程。嗯，这问题当年让我排查了整整一个下午。

4.3 进程间通信（IPC）

进程之间内存是隔离的，不能像线程那样共享全局变量。那怎么通信呢？multiprocessing提供了两种主流方式：Queue和Pipe。

4.3.1 Queue——生产者消费者模式

Queue是进程安全的队列，底层用了锁和管道。我个人习惯用它来做任务分发和结果收集。

from multiprocessing import Process, Queue

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

def consumer(q):
    while True:
        data = q.get()
        if data == "STOP":
            break
        print(f"消费: {data}")

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("STOP")  # 发送停止信号
    p2.join()

我的经验：Queue的get()方法是阻塞的。如果队列为空，它会一直等。生产者的速度远慢于消费者时，记得设置超时参数q.get(timeout=5)，避免死等。

4.3.2 Pipe——双向通信管道

Pipe返回一对连接对象，一个发一个收。它比Queue轻量，但只能用于两个进程之间。

from multiprocessing import Process, Pipe

def worker(conn):
    conn.send("你好，主进程")
    msg = conn.recv()
    print(f"子进程收到: {msg}")
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=worker, args=(child_conn,))
    p.start()
    
    msg = parent_conn.recv()
    print(f"主进程收到: {msg}")
    parent_conn.send("收到，子进程")
    
    p.join()

注意：Pipe不是线程安全的。如果多个线程同时读写同一个Pipe的端点，数据可能会乱。我曾经在一个项目中同时用Pipe和线程池，结果数据对不上，排查了好久才发现是并发读写的问题。多进程场景下，多个子进程共享一个Pipe端点，同样会有问题。

4.4 进程池（Pool）

手动创建和销毁进程开销很大。你想想看，每个进程都要复制一份内存空间、初始化解释器。如果任务数量多，频繁创建进程会拖慢整体性能。

进程池就是解决方案。它预先创建一批进程，任务来了直接分配，用完回收。我习惯用Pool来处理批量计算任务。

from multiprocessing import Pool
import time

def square(n):
    time.sleep(1)  # 模拟耗时计算
    return n * n

if __name__ == '__main__':
    numbers = list(range(10))
    
    # 创建4个进程的进程池
    with Pool(processes=4) as pool:
        # map: 阻塞等待所有结果
        results = pool.map(square, numbers)
        print(f"map结果: {results}")
        
        # apply_async: 异步提交
        async_results = [pool.apply_async(square, (n,)) for n in numbers]
        async_results = [r.get() for r in async_results]
        print(f"apply_async结果: {async_results}")

4.4.1 进程池的三种提交方式

方法 说明 适用场景
map() 阻塞等待所有结果，保持输入顺序 任务量已知，需要有序结果
apply_async() 异步提交，返回AsyncResult对象 任务耗时不同，需要非阻塞
imap() 惰性迭代，边计算边返回 数据量大，内存受限

进程池数量怎么定？

一般设置为CPU核心数。可以用os.cpu_count()获取。但要注意：如果任务涉及大量I/O等待，可以适当增加进程数。我一般会先设为核心数的1.5倍，然后压测调优。

4.5 避坑指南

做多进程开发，有几个坑我反复踩过，今天一并告诉你：

  • 全局变量不共享：每个进程都有独立的内存空间。修改全局变量不会影响其他进程。想共享数据？用Queue或Manager。
  • 文件描述符限制：每个进程都会占用文件描述符。进程数太多，系统会报"Too many open files"。我曾在一次爬虫项目中开了200个进程，结果系统直接崩溃。
  • 子进程异常处理：子进程崩溃不会影响主进程。但如果你不检查返回值，可能一直等不到结果。建议用join(timeout)加超时保护。
  • Windows下的特殊问题：Windows没有fork，创建进程的开销比Linux大很多。而且全局变量在子进程中会重新初始化。跨平台开发时，建议用multiprocessing.freeze_support()。

4.6 实战建议

最后，给你几个我自己的实践心得：

  1. 能用进程池就别手动管理进程：Pool帮你处理了创建、销毁、异常等琐事，代码更简洁。
  2. 进程间通信尽量用Queue：Pipe虽然轻量，但容易出并发问题。Queue封装得更好，适合大多数场景。
  3. 注意序列化开销：进程间传递的数据需要pickle序列化。大数据对象（比如大列表、DataFrame）的序列化时间可能比计算本身还长。这时候可以考虑共享内存（SharedMemory）。
  4. 先测试再上线：多进程的bug往往难以复现。我习惯先在本地用少量数据跑通，再逐步增加并发量。

多进程编程，说白了就是让Python真正利用多核CPU的能力。虽然有一些坑，但只要掌握了Queue、Pipe和Pool这三个核心工具，大部分场景都能应对。下一章，我们会聊聊协程——另一种更轻量的并发方案。

← 上一章 📖 返回目录 下一章 →