Python multiprocessing 实战指南

import multiprocessing
import os

def worker(name: str, count: int) -> None:
    """子进程执行的任务函数"""
    pid = os.getpid()
    print(f"[进程 {pid}] {name} 开始工作，计数: {count}")
    for i in range(count):
        print(f"[进程 {pid}] {name} 计数: {i + 1}/{count}")
    print(f"[进程 {pid}] {name} 工作完成")

if __name__ == "__main__":
    # 创建进程实例
    p1 = multiprocessing.Process(target=worker, args=("Worker-A", 3))
    p2 = multiprocessing.Process(target=worker, args=("Worker-B", 2))

    # 启动进程
    p1.start()
    p2.start()

    # 等待进程结束
    p1.join()
    p2.join()

    print("所有进程已完成")

import multiprocessing
import time

def background_monitor():
    """后台监控任务"""
    while True:
        print("监控中...")
        time.sleep(1)

if __name__ == "__main__":
    p = multiprocessing.Process(target=background_monitor)
    p.daemon = True  # 设置为守护进程
    p.start()

    # 主进程运行 3 秒后退出
    time.sleep(3)
    print("主进程退出，守护进程将自动终止")
    # 无需 p.join()，守护进程会随主进程退出

import multiprocessing as mp

# 查看当前启动方法
print(mp.get_start_method())

# 设置启动方法（必须在创建进程前）
mp.set_start_method('spawn')  # 或 'fork' 或 'forkserver'

import multiprocessing as mp

def worker():
    print("子进程运行中")

if __name__ == "__main__":
    # 跨平台推荐使用 spawn
    mp.set_start_method('spawn')

    p = mp.Process(target=worker)
    p.start()
    p.join()

from concurrent.futures import ProcessPoolExecutor
import os

def compute_square(n: int) -> int:
    """计算平方的 CPU 密集型任务"""
    pid = os.getpid()
    print(f"[进程 {pid}] 计算 {n}^2")
    return n * n

if __name__ == "__main__":
    # 创建进程池，默认使用 CPU 核心数
    with ProcessPoolExecutor(max_workers=4) as executor:
        # 提交单个任务
        future = executor.submit(compute_square, 10)
        result = future.result()
        print(f"结果: {result}")

from concurrent.futures import ProcessPoolExecutor, as_completed
import time

def slow_task(name: str, duration: float) -> str:
    """模拟耗时任务"""
    time.sleep(duration)
    return f"{name} 完成，耗时 {duration}s"

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=3) as executor:
        # 提交多个任务
        futures = {
            executor.submit(slow_task, "任务A", 2.0): "任务A",
            executor.submit(slow_task, "任务B", 1.0): "任务B",
            executor.submit(slow_task, "任务C", 3.0): "任务C",
        }

        # 按完成顺序获取结果
        for future in as_completed(futures):
            task_name = futures[future]
            result = future.result()
            print(f"{task_name}: {result}")

from concurrent.futures import ProcessPoolExecutor

def process_item(item: int) -> int:
    """处理单个数据项"""
    return item ** 2 + item * 2

if __name__ == "__main__":
    data = range(1, 11)

    with ProcessPoolExecutor(max_workers=4) as executor:
        # 并行处理，保持顺序
        results = list(executor.map(process_item, data))
        print(f"输入: {list(data)}")
        print(f"输出: {results}")

from concurrent.futures import ProcessPoolExecutor, as_completed
import random
import time

def random_task(n: int) -> tuple[int, float]:
    """随机耗时的任务"""
    delay = random.uniform(0.5, 2.0)
    time.sleep(delay)
    return n, delay

if __name__ == "__main__":
    with ProcessPoolExecutor(max_workers=4) as executor:
        futures = [executor.submit(random_task, i) for i in range(1, 6)]

        print("按完成顺序输出:")
        for future in as_completed(futures):
            n, delay = future.result()
            print(f"  任务 {n} 完成，耗时 {delay:.2f}s")

# ProcessPoolExecutor 推荐写法
from concurrent.futures import ProcessPoolExecutor

def task(n):
    return n * 2

if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        results = list(executor.map(task, range(10)))

import multiprocessing

def sender(conn):
    """发送端进程"""
    messages = ["Hello", "World", "Python", "Multiprocessing"]
    for msg in messages:
        print(f"[发送端] 发送: {msg}")
        conn.send(msg)
    conn.send(None)  # 发送结束信号
    conn.close()

def receiver(conn):
    """接收端进程"""
    while True:
        msg = conn.recv()
        if msg is None:
            break
        print(f"[接收端] 收到: {msg}")
    conn.close()

if __name__ == "__main__":
    # 创建管道，返回两个连接对象
    parent_conn, child_conn = multiprocessing.Pipe()

    # 创建进程
    p1 = multiprocessing.Process(target=sender, args=(parent_conn,))
    p2 = multiprocessing.Process(target=receiver, args=(child_conn,))

    p1.start()
    p2.start()

    p1.join()
    p2.join()

import multiprocessing
import time
import random

def producer(queue: multiprocessing.Queue, name: str, count: int):
    """生产者进程"""
    for i in range(count):
        item = f"{name}-商品{i}"
        queue.put(item)
        print(f"[生产者 {name}] 生产: {item}")
        time.sleep(random.uniform(0.1, 0.3))
    queue.put(None)  # 发送结束信号

def consumer(queue: multiprocessing.Queue, name: str):
    """消费者进程"""
    while True:
        item = queue.get()
        if item is None:
            print(f"[消费者 {name}] 收到结束信号，退出")
            break
        print(f"[消费者 {name}] 消费: {item}")
        time.sleep(random.uniform(0.2, 0.5))

if __name__ == "__main__":
    queue = multiprocessing.Queue()

    # 创建 2 个生产者和 2 个消费者
    producers = [
        multiprocessing.Process(target=producer, args=(queue, "P1", 5)),
        multiprocessing.Process(target=producer, args=(queue, "P2", 5)),
    ]
    consumers = [
        multiprocessing.Process(target=consumer, args=(queue, "C1")),
        multiprocessing.Process(target=consumer, args=(queue, "C2")),
    ]

    # 启动所有进程
    for p in producers + consumers:
        p.start()

    # 等待生产者完成
    for p in producers:
        p.join()

    # 发送额外的结束信号给每个消费者
    for _ in consumers:
        queue.put(None)

    # 等待消费者完成
    for c in consumers:
        c.join()

import multiprocessing
from multiprocessing import Manager

def worker(shared_dict: dict, shared_list: list, worker_id: int):
    """使用共享对象的进程"""
    # 修改共享字典
    shared_dict[f"worker_{worker_id}"] = f"数据来自进程 {worker_id}"

    # 修改共享列表
    shared_list.append(worker_id * 10)

    print(f"[Worker {worker_id}] 已写入数据")

if __name__ == "__main__":
    with Manager() as manager:
        # 创建共享数据结构
        shared_dict = manager.dict()
        shared_list = manager.list()

        # 创建多个进程
        processes = [
            multiprocessing.Process(target=worker, args=(shared_dict, shared_list, i))
            for i in range(4)
        ]

        for p in processes:
            p.start()

        for p in processes:
            p.join()

        # 主进程读取共享数据
        print(f"共享字典: {dict(shared_dict)}")
        print(f"共享列表: {list(shared_list)}")

import multiprocessing
from multiprocessing import Value, Array
import time

def increment_counter(counter: Value, iterations: int):
    """增加计数器"""
    for _ in range(iterations):
        with counter.get_lock():  # 获取锁
            counter.value += 1

if __name__ == "__main__":
    # 创建共享整数，初始值为 0
    counter = Value('i', 0)  # 'i' 表示有符号整数

    # 创建多个进程同时增加计数器
    processes = [
        multiprocessing.Process(target=increment_counter, args=(counter, 10000))
        for _ in range(4)
    ]

    for p in processes:
        p.start()

    for p in processes:
        p.join()

    print(f"最终计数: {counter.value}")  # 应该是 40000

import multiprocessing
from multiprocessing import Array
import math

def compute_partial_sum(arr: Array, start: int, end: int, result: Array, idx: int):
    """计算数组部分元素的和"""
    partial = sum(arr[start:end])
    result[idx] = partial

if __name__ == "__main__":
    # 创建共享数组
    data = list(range(1, 101))  # 1 到 100
    shared_arr = Array('i', data)  # 'i' 表示整数类型
    result_arr = Array('d', [0.0] * 4)  # 'd' 表示双精度浮点数

    # 分割任务
    chunk_size = len(data) // 4
    processes = []

    for i in range(4):
        start = i * chunk_size
        end = start + chunk_size if i < 3 else len(data)
        p = multiprocessing.Process(
            target=compute_partial_sum,
            args=(shared_arr, start, end, result_arr, i)
        )
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    # 汇总结果
    total = sum(result_arr)
    print(f"数组总和: {total}")  # 5050

import multiprocessing
from multiprocessing import Value, Array, Manager
import time

def benchmark_shared_memory():
    """对比共享内存和 Manager 的性能"""
    N = 1_000_000

    # 方式 1: Value/Array (共享内存)
    start = time.time()
    counter = Value('i', 0)

    def inc_value(c, n):
        for _ in range(n):
            with c.get_lock():
                c.value += 1

    p = multiprocessing.Process(target=inc_value, args=(counter, N))
    p.start()
    p.join()
    print(f"Value 耗时: {time.time() - start:.3f}s")

    # 方式 2: Manager (代理对象)
    start = time.time()
    with Manager() as manager:
        m_counter = manager.Value('i', 0)

        def inc_manager(c, n):
            for _ in range(n):
                c.value += 1

        p = multiprocessing.Process(target=inc_manager, args=(m_counter, N))
        p.start()
        p.join()
        print(f"Manager 耗时: {time.time() - start:.3f}s")

if __name__ == "__main__":
    benchmark_shared_memory()

# 多线程版本 - 受 GIL 限制
import threading
import time

def cpu_bound_task(n: int) -> int:
    """CPU 密集型任务"""
    return sum(i * i for i in range(n))

def threaded_version():
    start = time.time()
    threads = []
    for _ in range(4):
        t = threading.Thread(target=cpu_bound_task, args=(1000000,))
        threads.append(t)
        t.start()
    for t in threads:
        t.join()
    print(f"多线程耗时: {time.time() - start:.2f}s")

# 多进程版本 - 真正并行
import multiprocessing

def process_version():
    start = time.time()
    processes = []
    for _ in range(4):
        p = multiprocessing.Process(target=cpu_bound_task, args=(1000000,))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()
    print(f"多进程耗时: {time.time() - start:.2f}s")

if __name__ == "__main__":
    threaded_version()
    process_version()

from concurrent.futures import ProcessPoolExecutor, as_completed
from collections import Counter
import re
from pathlib import Path

def process_chunk(file_path: str, start: int, end: int) -> Counter:
    """处理文件的一个数据块"""
    word_counts = Counter()
    with open(file_path, 'r', encoding='utf-8') as f:
        f.seek(start)
        # 确保从完整的行开始
        if start > 0:
            f.readline()
        lines = []
        pos = f.tell()
        while pos < end:
            line = f.readline()
            if not line:
                break
            lines.append(line)
            pos = f.tell()

        # 统计词频
        for line in lines:
            words = re.findall(r'\b\w+\b', line.lower())
            word_counts.update(words)

    return word_counts

def parallel_word_count(file_path: str, num_workers: int = 4) -> Counter:
    """并行统计大文件词频"""
    file_size = Path(file_path).stat().st_size
    chunk_size = file_size // num_workers

    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        futures = []
        for i in range(num_workers):
            start = i * chunk_size
            end = start + chunk_size if i < num_workers - 1 else file_size
            future = executor.submit(process_chunk, file_path, start, end)
            futures.append(future)

        # 合并结果
        total_counts = Counter()
        for future in as_completed(futures):
            total_counts.update(future.result())

    return total_counts

if __name__ == "__main__":
    # 示例：统计大文件词频
    # result = parallel_word_count("large_file.txt", num_workers=8)
    # print(f"最常见的 10 个词: {result.most_common(10)}")
    pass

from concurrent.futures import ProcessPoolExecutor
from PIL import Image, ImageFilter
from pathlib import Path
import time

def process_image(image_path: str, output_dir: str, operations: list) -> str:
    """处理单张图片"""
    img = Image.open(image_path)

    for op in operations:
        if op == 'grayscale':
            img = img.convert('L')
        elif op == 'blur':
            img = img.filter(ImageFilter.GaussianBlur(radius=2))
        elif op == 'resize':
            img = img.resize((800, 600))
        elif op == 'thumbnail':
            img.thumbnail((200, 200))

    # 保存处理后的图片
    output_path = Path(output_dir) / Path(image_path).name
    img.save(output_path)
    return str(output_path)

def batch_process_images(input_dir: str, output_dir: str, num_workers: int = 4):
    """批量处理图片"""
    input_path = Path(input_dir)
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    # 获取所有图片文件
    image_files = list(input_path.glob("*.jpg")) + list(input_path.glob("*.png"))

    operations = ['grayscale', 'blur', 'thumbnail']

    start = time.time()

    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        futures = {
            executor.submit(process_image, str(img), str(output_path), operations): img
            for img in image_files
        }

        completed = 0
        for future in as_completed(futures):
            completed += 1
            result = future.result()
            print(f"[{completed}/{len(image_files)}] 完成: {result}")

    print(f"总耗时: {time.time() - start:.2f}s")

if __name__ == "__main__":
    # batch_process_images("input_images", "output_images", num_workers=8)
    pass

from concurrent.futures import ProcessPoolExecutor
import math
import time

def is_prime(n: int) -> bool:
    """判断是否为质数"""
    if n < 2:
        return False
    if n == 2:
        return True
    if n % 2 == 0:
        return False
    for i in range(3, int(math.sqrt(n)) + 1, 2):
        if n % i == 0:
            return False
    return True

def find_primes_in_range(start: int, end: int) -> list[int]:
    """在指定范围内查找质数"""
    return [n for n in range(start, end) if is_prime(n)]

def parallel_prime_search(limit: int, num_workers: int = 4) -> list[int]:
    """并行查找质数"""
    chunk_size = limit // num_workers
    ranges = [
        (i * chunk_size, (i + 1) * chunk_size if i < num_workers - 1 else limit)
        for i in range(num_workers)
    ]

    start_time = time.time()

    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        futures = [
            executor.submit(find_primes_in_range, r[0], r[1])
            for r in ranges
        ]

        all_primes = []
        for future in futures:
            all_primes.extend(future.result())

    all_primes.sort()
    elapsed = time.time() - start_time
    print(f"在 1-{limit} 范围内找到 {len(all_primes)} 个质数")
    print(f"耗时: {elapsed:.2f}s")

    return all_primes

if __name__ == "__main__":
    # 查找 1 到 1000000 的所有质数
    primes = parallel_prime_search(1_000_000, num_workers=8)

# 错误示例 - Windows 下会无限创建进程
import multiprocessing

def worker():
    print("工作进程")

# 错误：没有保护，Windows 下会递归创建进程
p = multiprocessing.Process(target=worker)
p.start()
p.join()

# 正确示例
import multiprocessing

def worker():
    print("工作进程")

if __name__ == "__main__":
    p = multiprocessing.Process(target=worker)
    p.start()
    p.join()

# 错误示例 - 没有锁保护
from multiprocessing import Value

def unsafe_increment(counter: Value, n: int):
    for _ in range(n):
        counter.value += 1  # 非原子操作！

if __name__ == "__main__":
    counter = Value('i', 0)
    # 多进程并发增加，结果可能不正确

# 正确示例 - 使用锁保护
from multiprocessing import Value

def safe_increment(counter: Value, n: int):
    for _ in range(n):
        with counter.get_lock():  # 获取锁
            counter.value += 1

if __name__ == "__main__":
    counter = Value('i', 0)
    # 结果正确

# 1. 合理设置进程数
import os
# CPU 密集型：进程数 = CPU 核心数
max_workers = os.cpu_count()

# I/O 密集型：进程数可以更多
max_workers = os.cpu_count() * 2

# 2. 批量处理减少 IPC 开销
def process_batch(items: list) -> list:
    """批量处理，减少进程间通信次数"""
    return [process_item(item) for item in items]

# 3. 使用共享内存避免序列化
from multiprocessing import Array
shared_data = Array('d', large_data)  # 避免传递大列表

# 4. 任务分块
def chunked_map(executor, func, data, chunk_size=1000):
    """分块处理大数据集"""
    chunks = [data[i:i+chunk_size] for i in range(0, len(data), chunk_size)]
    results = executor.map(process_batch, chunks)
    return [item for chunk in results for item in chunk]