Python 多进程实战：multiprocessing 入门

发布于 约 13 分钟

Python 的 GIL 让多线程无法真正并行计算，CPU 密集型任务要用多进程。本文从进程创建、进程间通信到进程池，讲清 multiprocessing 的核心用法。

Python 多进程配图：蓝色渐变背景上多个并行工作节点共享中央任务分配面板，闪电连接，扁平科技插画
multiprocessing：绕过 GIL 的真正并行计算

一、为什么需要多进程

Python 有 GIL（全局解释器锁），同一时刻只有一个线程执行 Python 字节码。多线程适合 IO 密集任务（网络请求、文件读写），但 CPU 密集任务（数值计算、数据处理）要用多进程。

类型多线程多进程
CPU 密集受 GIL 限制，不快真正并行，快
IO 密集合适合适但开销大
内存共享各自独立
启动开销小大

二、创建第一个进程

from multiprocessing import Process
import time

def worker(name):
    print(f"进程 {name} 开始工作")
    time.sleep(2)
    print(f"进程 {name} 完成")

if __name__ == "__main__":
    p1 = Process(target=worker, args=("A",))
    p2 = Process(target=worker, args=("B",))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    print("主进程结束")
注意：Windows 下多进程必须放在 `if __name__ == "__main__":` 里，否则子进程会无限递归创建进程。这是 Windows 的 spawn 启动方式决定的。

三、进程池

from multiprocessing import Pool

def square(x):
    return x * x

if __name__ == "__main__":
    with Pool(processes=4) as pool:
        # map：批量映射
        results = pool.map(square, range(10))
        print(results)

        # async：异步获取结果
        async_result = pool.apply_async(square, (5,))
        print(async_result.get())

        # imap：迭代器式返回
        for r in pool.imap(square, range(5)):
            print(r)

四、进程间通信

from multiprocessing import Process, Queue

def producer(q):
    for i in range(5):
        q.put(f"任务 {i}")

def consumer(q):
    while True:
        item = q.get()
        if item is None:
            break
        print(f"消费：{item}")

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()

五、共享内存

from multiprocessing import Process, Value, Array

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

if __name__ == "__main__":
    counter = Value("i", 0)  # 共享整数
    p1 = Process(target=increment, args=(counter,))
    p2 = Process(target=increment, args=(counter,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    print(f"最终值：{counter.value}")

六、实战：并行处理数据

from multiprocessing import Pool
import time

def process_chunk(chunk_id):
    # 模拟 CPU 密集任务
    total = sum(i * i for i in range(1_000_000))
    return f"chunk {chunk_id}: {total}"

if __name__ == "__main__":
    start = time.time()
    with Pool(processes=4) as pool:
        results = pool.map(process_chunk, range(8))
    elapsed = time.time() - start
    for r in results:
        print(r)
    print(f"耗时：{elapsed:.2f} 秒")

七、总结

CPU 密集任务用多进程，IO 密集任务用多线程或异步。multiprocessing 的核心是 Process（创建进程）、Pool（进程池）、Queue（通信）、Value/Array（共享内存）。记住 Windows 下必须加 `if __name__ == "__main__":`，进程间数据默认隔离，通信要靠 Queue 或共享内存。

延伸阅读