1. 多进程 (multiprocessing 模块)
   基本使用

import multiprocessing
import time

def cpu_intensive_task(n, process_id):
    """CPU 密集型任务"""
    result = 0
    for i in range(n):
        result += i * i
    print(f"进程 {process_id} 完成，结果: {result}")
    return result

if __name__ == "__main__":
    start_time = time.time()
    
    # 创建进程
    processes = []
    for i in range(4):
        p = multiprocessing.Process(
            target=cpu_intensive_task, 
            args=(10000000, i)
        )
        processes.append(p)
        p.start()
    
    # 等待所有进程完成
    for p in processes:
        p.join()
    
    end_time = time.time()
    print(f"多进程执行时间: {end_time - start_time:.2f} 秒")

进程池

import multiprocessing
import os

def square(x):
    """计算平方"""
    print(f"进程 {os.getpid()} 计算 {x} 的平方")
    return x * x

if __name__ == "__main__":
    numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    
    # 使用进程池
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(square, numbers)
    
    print(f"结果: {results}")

进程间通信
使用 Queue

import multiprocessing
import time

def producer(queue, items):
    """生产者进程"""
    for item in items:
        print(f"生产: {item}")
        queue.put(item)
        time.sleep(0.5)
    queue.put(None)  # 结束信号

def consumer(queue):
    """消费者进程"""
    while True:
        item = queue.get()
        if item is None:  # 收到结束信号
            break
        print(f"消费: {item}")
        time.sleep(0.5)

if __name__ == "__main__":
    # 创建进程安全的队列
    queue = multiprocessing.Queue()
    
    # 创建进程
    p1 = multiprocessing.Process(
        target=producer, 
        args=(queue, ['A', 'B', 'C', 'D', 'E'])
    )
    p2 = multiprocessing.Process(target=consumer, args=(queue,))
    
    # 启动进程
    p1.start()
    p2.start()
    
    # 等待结束
    p1.join()
    p2.join()

使用 Pipe

import multiprocessing

def sender(conn, messages):
    """发送消息的进程"""
    for message in messages:
        print(f"发送: {message}")
        conn.send(message)
        response = conn.recv()  # 等待回应
        print(f"收到回应: {response}")
    conn.close()

def receiver(conn):
    """接收消息的进程"""
    while True:
        try:
            message = conn.recv()
            print(f"接收: {message}")
            conn.send(f"已收到: {message}")
        except EOFError:
            break
    conn.close()

if __name__ == "__main__":
    # 创建管道
    parent_conn, child_conn = multiprocessing.Pipe()
    
    messages = ['Hello', 'World', 'Python', 'Multiprocessing']
    
    p1 = multiprocessing.Process(target=sender, args=(parent_conn, messages))
    p2 = multiprocessing.Process(target=receiver, args=(child_conn,))
    
    p1.start()
    p2.start()
    
    p1.join()
    p2.join()

共享内存

import multiprocessing

def worker(shared_value, shared_array, lock, worker_id):
    """修改共享数据的进程"""
    for i in range(5):
        with lock:  # 使用锁保证数据安全
            shared_value.value += 1
            shared_array[worker_id] += 1
        print(f"Worker {worker_id}: value={shared_value.value}, array={list(shared_array)}")

if __name__ == "__main__":
    # 创建共享值
    shared_value = multiprocessing.Value('i', 0)  # 'i' 表示整数
    
    # 创建共享数组
    shared_array = multiprocessing.Array('i', [0, 0, 0])  # 长度为3的整数数组
    
    # 创建锁
    lock = multiprocessing.Lock()
    
    # 创建多个进程
    processes = []
    for i in range(3):
        p = multiprocessing.Process(
            target=worker, 
            args=(shared_value, shared_array, lock, i)
        )
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()
    
    print(f"最终结果: value={shared_value.value}, array={list(shared_array)}")

2. 现代用法：concurrent.futures 高级接口
   ThreadPoolExecutor 和 ProcessPoolExecutor 的统一接口

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
import math

def is_prime(n):
    """判断是否为质数（CPU密集型）"""
    if n < 2:
        return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0:
            return False
    return True

def check_url(url):
    """检查URL（IO密集型）"""
    time.sleep(1)  # 模拟网络请求
    return f"{url} - OK"

# 根据任务类型选择执行器
def run_tasks(use_processes=False):
    numbers = [112272535095293, 112582705942171, 115280095190773, 115797848077099]
    urls = ['http://example.com/1', 'http://example.com/2', 'http://example.com/3']
    
    if use_processes:
        # CPU密集型使用进程池
        executor_class = ProcessPoolExecutor
        tasks = [(is_prime, n) for n in numbers]
    else:
        # IO密集型使用线程池
        executor_class = ThreadPoolExecutor  
        tasks = [(check_url, url) for url in urls]
    
    with executor_class(max_workers=4) as executor:
        # 使用 submit
        futures = [executor.submit(func, arg) for func, arg in tasks]
        
        # 使用 as_completed 按完成顺序获取结果
        for future in as_completed(futures):
            try:
                result = future.result()
                print(f"完成: {result}")
            except Exception as e:
                print(f"错误: {e}")

if __name__ == "__main__":
    print("=== 线程池（IO密集型）===")
    run_tasks(use_processes=False)
    
    print("\n=== 进程池（CPU密集型）===")
    run_tasks(use_processes=True)