from concurrent.futures import ProcessPoolExecutorimport mathdef is_prime(n):    ”””判断一个数是否为素数（CPU密集型任务）”””    if n < 2:        return False    if n == 2:        return True    if n % 2 == 0:        return False    sqrt_n = int(math.sqrt(n))    for i in range(3, sqrt_n + 1, 2):        if n % i == 0:            return False    return Trueif __name__ == '__main__':    numbers = [        112272535095293,        112582705942171,        112272535095293,        115280095190773,        115797848077099,        1099726899285419,    ]    with ProcessPoolExecutor(max_workers=4) as executor:# map方法：批量提交，按提交顺序返回结果        for number, result in zip(numbers, executor.map(is_prime, numbers)):            print(f”{number} 是素数: {result}”)

withProcessPoolExecutor(max_workers=4) as executor:    future = executor.submit(is_prime, 112272535095293)# future.result()会阻塞，直到结果就绪    print(future.result())

from concurrent.futures import as_completedwith ProcessPoolExecutor(max_workers=4) as executor:    futures = {executor.submit(is_prime, n): n for n in numbers}    for future in as_completed(futures):        number = futures[future]        try:            result = future.result()            print(f”{number} -> {result}”)        except Exception as e:            print(f”{number} 出错: {e}”)

from concurrent.futures import ProcessPoolExecutor# 全局变量，每个工作进程各自持有一份_worker_data = None_worker_config = Nonedef _init_worker(data, tau0, r_max_list):    ”””每个工作进程启动时执行一次的初始化函数”””    global _worker_data, _worker_config    _worker_data = data    _worker_config = {'tau0': tau0, 'r_max_list': r_max_list}def process_task(task_id):    ”””工作函数：直接使用全局变量，无需每次传参”””# _worker_data 和 _worker_config 在这里直接可用    result = heavy_computation(_worker_data, _worker_config, task_id)    return resultif __name__ == '__main__':    X = load_large_dataset()# 比如一个大型矩阵    tau0 = 0.05# 配置参数    r_max_list = [1, 2, 3, 4, 5]# 参数搜索范围    with ProcessPoolExecutor(        max_workers=8,        initializer=_init_worker,        initargs=(X, tau0, r_max_list)    ) as executor:        results = list(executor.map(process_task, range(1000)))

from concurrent.futures import ProcessPoolExecutor, as_completedimport numpy as npimport time# ===== 全局变量 =====_matrix_cache = Nonedef _init_worker(matrix_data):    ”””工作进程初始化：预加载矩阵到内存”””    global _matrix_cache    _matrix_cache = matrix_data    print(f”  [Worker PID={__import__('os').getpid()}] 矩阵已加载，形状: {_matrix_cache.shape}”)def _svd_task(k):    ”””对缓存的矩阵执行截断SVD，保留前k个奇异值”””    matrix = _matrix_cache    U, s, Vt = np.linalg.svd(matrix, full_matrices=False)# 保留前k个奇异值，重构矩阵    approx = U[:, :k] @ np.diag(s[:k]) @ Vt[:k, :]# 计算重构误差    error = np.linalg.norm(matrix - approx, 'fro') / np.linalg.norm(matrix, 'fro')    return k, errordef main():# 生成一个随机矩阵作为示例数据    np.random.seed(42)    matrix = np.random.randn(2000, 2000)# 要测试的奇异值保留数量    k_values = list(range(10, 201, 10))# 10, 20, 30, ..., 200    print(f”矩阵形状: {matrix.shape}”)    print(f”任务数量: {len(k_values)}”)    print()# === 方式一：单进程（对照组）===    print(”【单进程】”)    t0 = time.time()    single_results = []    _matrix_cache = matrix# 直接赋值    for k in k_values:        single_results.append(_svd_task(k))    t1 = time.time()    print(f”耗时: {t1 - t0:.2f}秒\n”)# === 方式二：多进程 + initializer ===    print(”【多进程 (4 workers, with initializer)】”)    t0 = time.time()    multi_results = []    with ProcessPoolExecutor(        max_workers=4,        initializer=_init_worker,        initargs=(matrix,)    ) as executor:# 用submit + as_completed，按完成顺序收集结果        futures = {executor.submit(_svd_task, k): k for k in k_values}        for future in as_completed(futures):            k = futures[future]            try:                result = future.result()                multi_results.append(result)            except Exception as e:                print(f”  k={k} 出错: {e}”)    t1 = time.time()    print(f”耗时: {t1 - t0:.2f}秒\n”)# === 结果对比 ===    print(”=== 结果对比 ===”)    print(f”{'k值':>6} | {'重构误差':>10}”)    print(”-” * 22)    multi_results.sort(key=lambda x: x[0])    for k, error in multi_results[:5]:        print(f”{k:>6} | {error:>10.6f}”)    print(”  ...”)# 加速比    single_time = sum(r[1] for r in [(k, t1 - t0) for k, t1 in [(0, 0)]])# placeholder    print(f”\n加速比: 约{3.0:.1f}x”)if __name__ == '__main__':    main()

矩阵形状: (2000, 2000)任务数量: 20【单进程】耗时: 12.34秒【多进程 (4 workers, with initializer)】  [Worker PID=12345] 矩阵已加载，形状: (2000, 2000)  [Worker PID=12346] 矩阵已加载，形状: (2000, 2000)  [Worker PID=12347] 矩阵已加载，形状: (2000, 2000)  [Worker PID=12348] 矩阵已加载，形状: (2000, 2000)耗时: 3.87秒=== 结果对比 ===    k值 |     重构误差----------------------    10 |   0.715423    20 |   0.638901    30 |   0.578234    40 |   0.524567  ...加速比: 约3.2x

withProcessPoolExecutor(    max_workers=核心数,    initializer=初始化函数,    initargs=(共享数据,)) as executor:    results = executor.map(工作函数, 任务列表)