多进程并发

多进程通过创建独立的 Python 解释器进程来实现真正的并行执行，彻底绕过 GIL 的限制。每个进程有自己的内存空间和 GIL，适合 CPU 密集型任务。

进程 vs 线程

内存隔离

进程间不共享内存，通信需要显式传递数据：

import multiprocessing
import threading

shared_value = 0

def modify_in_thread():
    global shared_value
    shared_value = 42

def modify_in_process():
    global shared_value
    shared_value = 42

# 线程修改：影响主进程
t = threading.Thread(target=modify_in_thread)
t.start()
t.join()
print(f"线程修改后: {shared_value}")  # 输出: 42

shared_value = 0  # 重置

# 进程修改：不影响主进程
p = multiprocessing.Process(target=modify_in_process)
p.start()
p.join()
print(f"进程修改后: {shared_value}")  # 输出: 0（未改变）

GIL 绕过

多进程真正实现并行计算：

import multiprocessing
import time

def compute_primes(n):
    """计算前 n 个素数"""
    primes = []
    num = 2
    while len(primes) < n:
        is_prime = True
        for p in primes:
            if p * p > num:
                break
            if num % p == 0:
                is_prime = False
                break
        if is_prime:
            primes.append(num)
        num += 1
    return len(primes)

if __name__ == '__main__':
    # 单进程
    start = time.time()
    compute_primes(10000)
    single_time = time.time() - start
    print(f"单进程: {single_time:.2f}秒")
    
    # 多进程
    start = time.time()
    processes = []
    for _ in range(2):
        p = multiprocessing.Process(target=compute_primes, args=(5000,))
        p.start()
        processes.append(p)
    
    for p in processes:
        p.join()
    
    multi_time = time.time() - start
    print(f"双进程: {multi_time:.2f}秒")
    print(f"加速比: {single_time / multi_time:.2f}x")

输出示例（4核 CPU）：

单进程: 2.45秒
双进程: 1.28秒
加速比: 1.91x

注意：if __name__ == '__main__': 在 Windows/macOS 上是必须的，防止无限递归创建进程。

开销对比

特性	线程	进程
创建开销	低（~1ms）	高（~10-100ms）
内存占用	共享主进程内存	每个进程独立内存
通信方式	直接访问共享变量	IPC（Queue、Pipe）
GIL 影响	受限	无影响
适用场景	I/O 密集	CPU 密集

Process 类

基本用法

import multiprocessing
import os

def worker(name):
    print(f"{name} 运行在进程 {os.getpid()}")
    print(f"父进程 PID: {os.getppid()}")

if __name__ == '__main__':
    print(f"主进程 PID: {os.getpid()}")
    
    p = multiprocessing.Process(target=worker, args=("子进程",))
    p.start()
    p.join()
    
    print(f"子进程退出码: {p.exitcode}")  # 0 表示成功

进程生命周期

import multiprocessing
import time

def task():
    time.sleep(2)

if __name__ == '__main__':
    p = multiprocessing.Process(target=task)
    
    print(f"启动前: alive={p.is_alive()}, pid={p.pid}")
    p.start()
    print(f"启动后: alive={p.is_alive()}, pid={p.pid}")
    
    p.join(timeout=1)  # 等待1秒
    if p.is_alive():
        print("进程仍在运行")
        p.join()  # 继续等待
    
    print(f"结束后: alive={p.is_alive()}, exitcode={p.exitcode}")

终止进程

import multiprocessing
import time

def long_task():
    for i in range(100):
        print(f"步骤 {i}")
        time.sleep(0.1)

if __name__ == '__main__':
    p = multiprocessing.Process(target=long_task)
    p.start()
    
    time.sleep(1)
    
    # 优雅终止
    p.terminate()  # 发送 SIGTERM
    p.join()  # 等待清理
    
    print(f"退出码: {p.exitcode}")  # 负数表示被信号终止
    
    # 强制杀死（不推荐）
    # p.kill()  # 发送 SIGKILL，无法清理资源

进程池（Pool）

处理大量任务时，重复创建进程开销大。进程池预先创建固定数量的工作进程，重复使用它们处理任务。

map：批量处理

import multiprocessing
import math

def compute_factorial(n):
    return math.factorial(n)

if __name__ == '__main__':
    numbers = range(1, 11)
    
    # 顺序执行
    import time
    start = time.time()
    results_seq = [compute_factorial(n) for n in numbers]
    seq_time = time.time() - start
    
    # 并行执行
    start = time.time()
    with multiprocessing.Pool(processes=4) as pool:
        results_par = pool.map(compute_factorial, numbers)
    par_time = time.time() - start
    
    print(f"顺序: {seq_time:.4f}秒")
    print(f"并行: {par_time:.4f}秒")
    print(f"结果一致: {results_seq == results_par}")

map 保持输入顺序，阻塞直到所有任务完成。

apply_async：异步提交

import multiprocessing
import time

def slow_square(x):
    time.sleep(1)
    return x * x

if __name__ == '__main__':
    with multiprocessing.Pool(processes=2) as pool:
        # 异步提交多个任务
        results = []
        for i in range(4):
            result = pool.apply_async(slow_square, (i,))
            results.append(result)
        
        # 继续其他工作...
        print("任务已提交，主进程继续")
        
        # 获取结果
        for i, result in enumerate(results):
            print(f"{i}² = {result.get()}")  # get() 阻塞直到结果就绪

imap：迭代式处理

import multiprocessing

def process_item(x):
    return x * 2

if __name__ == '__main__':
    data = range(10)
    
    with multiprocessing.Pool(processes=4) as pool:
        # imap 返回迭代器，按完成顺序产出结果
        for result in pool.imap_unordered(process_item, data):
            print(f"完成: {result}")  # 结果顺序不确定

imap 适合大数据集，不会一次性加载所有任务到内存。

进程池上下文管理

if __name__ == '__main__':
    # 推荐方式：自动清理
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(lambda x: x**2, range(10))
    # 退出时自动 close() 和 join()
    
    # 手动管理
    pool = multiprocessing.Pool(processes=4)
    try:
        results = pool.map(lambda x: x**2, range(10))
    finally:
        pool.close()  # 不再接受新任务
        pool.join()   # 等待所有任务完成

进程间通信

Queue：队列

跨进程的线程安全队列：

import multiprocessing

def producer(queue, n):
    for i in range(n):
        queue.put(f"Item-{i}")
    queue.put(None)  # 哨兵值

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

if __name__ == '__main__':
    queue = multiprocessing.Queue()
    
    p1 = multiprocessing.Process(target=producer, args=(queue, 5))
    p2 = multiprocessing.Process(target=consumer, args=(queue,))
    
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()

Pipe：管道

双向通信通道：

import multiprocessing

def sender(conn):
    conn.send("Hello from child")
    response = conn.recv()
    print(f"子进程收到: {response}")
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = multiprocessing.Pipe()
    
    p = multiprocessing.Process(target=sender, args=(child_conn,))
    p.start()
    
    message = parent_conn.recv()
    print(f"父进程收到: {message}")
    parent_conn.send("Hello from parent")
    
    p.join()
    parent_conn.close()

Pipe vs Queue:

• Pipe 更快，但只能两个进程间通信
• Queue 支持多个生产者和消费者

Manager：共享复杂对象

import multiprocessing

def worker(shared_dict, shared_list, key, value):
    shared_dict[key] = value
    shared_list.append(value)

if __name__ == '__main__':
    with multiprocessing.Manager() as manager:
        shared_dict = manager.dict()
        shared_list = manager.list()
        
        processes = []
        for i in range(5):
            p = multiprocessing.Process(
                target=worker,
                args=(shared_dict, shared_list, f"key{i}", i * 10)
            )
            p.start()
            processes.append(p)
        
        for p in processes:
            p.join()
        
        print(f"字典: {dict(shared_dict)}")
        print(f"列表: {list(shared_list)}")

输出示例：

字典: {'key0': 0, 'key1': 10, 'key2': 20, 'key3': 30, 'key4': 40}
列表: [0, 10, 20, 30, 40]

注意：Manager 通过进程间代理实现，性能低于 Value/Array 的共享内存。

共享内存

Value 和 Array

直接在进程间共享内存：

import multiprocessing

def increment(shared_value, shared_array, index):
    shared_value.value += 1
    shared_array[index] = shared_value.value

if __name__ == '__main__':
    # 'd' 表示 double 类型
    shared_value = multiprocessing.Value('i', 0)  # 'i' 表示 int
    shared_array = multiprocessing.Array('i', 5)  # 5个整数的数组
    
    processes = []
    for i in range(5):
        p = multiprocessing.Process(
            target=increment,
            args=(shared_value, shared_array, i)
        )
        p.start()
        processes.append(p)
    
    for p in processes:
        p.join()
    
    print(f"最终值: {shared_value.value}")
    print(f"数组: {list(shared_array)}")

同步共享内存

共享内存访问需要同步：

import multiprocessing

def safe_increment(shared_value, lock):
    for _ in range(10000):
        with lock:
            shared_value.value += 1

if __name__ == '__main__':
    shared_value = multiprocessing.Value('i', 0)
    lock = multiprocessing.Lock()
    
    processes = [
        multiprocessing.Process(target=safe_increment, args=(shared_value, lock))
        for _ in range(4)
    ]
    
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    
    print(f"最终值: {shared_value.value}")  # 应该是 40000

实战案例：并行图像处理

import multiprocessing
from PIL import Image
import os

def process_image(input_path, output_path):
    """将图像转为灰度"""
    try:
        img = Image.open(input_path)
        gray_img = img.convert('L')
        gray_img.save(output_path)
        return f"成功: {os.path.basename(input_path)}"
    except Exception as e:
        return f"失败: {os.path.basename(input_path)} - {e}"

def process_batch(file_list):
    """批量处理图像"""
    results = []
    for input_file, output_file in file_list:
        result = process_image(input_file, output_file)
        results.append(result)
    return results

if __name__ == '__main__':
    # 假设有100张图像
    image_files = [f"input/image_{i}.jpg" for i in range(100)]
    output_files = [f"output/gray_{i}.jpg" for i in range(100)]
    
    tasks = list(zip(image_files, output_files))
    
    # 将任务分成4批
    import time
    start = time.time()
    
    num_processes = 4
    chunk_size = len(tasks) // num_processes
    chunks = [tasks[i:i+chunk_size] for i in range(0, len(tasks), chunk_size)]
    
    with multiprocessing.Pool(processes=num_processes) as pool:
        results = pool.map(process_batch, chunks)
    
    elapsed = time.time() - start
    
    # 展平结果
    all_results = [r for batch in results for r in batch]
    successful = sum(1 for r in all_results if r.startswith("成功"))
    
    print(f"处理完成: {successful}/{len(tasks)} 成功")
    print(f"耗时: {elapsed:.2f}秒")

性能对比：threading vs multiprocessing

import threading
import multiprocessing
import time

def cpu_bound_task(n):
    """CPU 密集型任务"""
    total = 0
    for i in range(n):
        total += i ** 2
    return total

def io_bound_task(n):
    """I/O 密集型任务"""
    time.sleep(n)
    return n

def benchmark(task_func, use_threads, num_workers, *args):
    start = time.time()
    
    if use_threads:
        workers = [threading.Thread(target=task_func, args=args) 
                   for _ in range(num_workers)]
    else:
        workers = [multiprocessing.Process(target=task_func, args=args) 
                   for _ in range(num_workers)]
    
    for w in workers:
        w.start()
    for w in workers:
        w.join()
    
    return time.time() - start

if __name__ == '__main__':
    print("=== CPU 密集型任务 ===")
    print(f"线程: {benchmark(cpu_bound_task, True, 4, 1000000):.2f}秒")
    print(f"进程: {benchmark(cpu_bound_task, False, 4, 1000000):.2f}秒")
    
    print("\n=== I/O 密集型任务 ===")
    print(f"线程: {benchmark(io_bound_task, True, 4, 1):.2f}秒")
    print(f"进程: {benchmark(io_bound_task, False, 4, 1):.2f}秒")

输出示例：

=== CPU 密集型任务 ===
线程: 3.45秒
进程: 0.92秒

=== I/O 密集型任务 ===
线程: 1.01秒
进程: 1.05秒

小结

• 多进程 通过独立内存空间实现真正的并行
• 进程池 避免重复创建进程的开销
• Queue/Pipe 用于进程间通信
• Value/Array 提供高性能的共享内存
• Manager 支持复杂对象共享，但性能较低
• CPU 密集型任务 选择多进程，I/O 密集型任务 选择多线程

下一节学习 asyncio，它用协作式调度实现高效的异步 I/O。

---

练习：答案