嵌套依赖循环场景下Python多进程最佳实践咨询

核心思路 #

1. 用进程池固定工作进程数量，避免无限制创建进程耗尽系统资源
2. 让generator1持续产出num1，每产出一个就给它分配指定数量的generator2任务（搭配不同num2），异步提交到进程池
3. 实时监听任务结果，一旦找到符合要求的目标，立刻终止所有进程并退出程序
4. 每个generator2任务内部持续计算，直到找到目标或被强制终止

具体代码实现 #

先对生成器做适配改造，让它能配合进程池异步任务逻辑：

import multiprocessing as mp
from random import random
import time

def generator1():
    # 模拟持续产出num1的无限生成器
    while True:
        yield random()

def generator2_task(num1, target_condition):
    # 单个任务：绑定num1，持续生成num2并计算结果
    while True:
        num2 = random()
        result = hash(str(num1**num2))
        # 检查是否满足目标条件，这里用结果小于指定阈值做示例
        if result < target_condition:
            return (num1, num2, result)
        # 非必要：加微延迟降低CPU占用，实际重计算场景可删除
        time.sleep(0.001)

def main():
    # 可配置参数
    WORKER_PROCESSES = 4  # 进程池总大小
    TARGET_THRESHOLD = -10**18  # 示例目标条件
    TASKS_PER_NUM1 = 2  # 每个num1分配的generator2任务数

    # 创建进程池，设置maxtasksperchild避免长期运行内存泄漏
    pool = mp.Pool(processes=WORKER_PROCESSES, maxtasksperchild=100)
    # 共享布尔变量，标记是否找到目标结果
    found_target = mp.Value('b', False)

    def on_result_found(result):
        # 任务结果回调函数：找到目标后立即终止流程
        nonlocal found_target
        if not found_target.value:
            print(f"找到目标：num1={result[0]}, num2={result[1]}, 计算结果={result[2]}")
            found_target.value = True
            # 强制终止所有进程，停止所有计算
            pool.terminate()

    try:
        for num1 in generator1():
            if found_target.value:
                break
            # 为当前num1提交指定数量的异步任务
            for _ in range(TASKS_PER_NUM1):
                if found_target.value:
                    break
                pool.apply_async(
                    generator2_task,
                    args=(num1, TARGET_THRESHOLD),
                    callback=on_result_found
                )
            # 可选：控制num1产出速度，避免任务堆积过多占用内存
            time.sleep(0.1)
    finally:
        # 收尾：关闭进程池并等待剩余任务（如果未提前终止）
        pool.close()
        pool.join()

if __name__ == '__main__':
    main()

方案优势 #

• 动态任务提交：apply_async支持在generator1产出num1的同时立即提交任务，完全符合你"产出第一个结果就启动对应进程"的需求，无需等待之前的任务完成
• 资源可控：固定进程池大小，避免手动创建独立进程时的资源失控问题
• 快速终止：通过共享变量和pool.terminate()，找到目标后能立刻停止所有计算，不做无效消耗
• 灵活适配：可以通过调整WORKER_PROCESSES和TASKS_PER_NUM1，根据硬件资源和业务需求优化并发效率

替代方案对比 #

• 手动创建独立进程：需要自己管理进程的创建、销毁、结果收集，代码复杂度高，容易出现进程泄漏、状态不同步等问题，不推荐
• imap/imap_unordered：这两个方法需要提前准备好所有任务参数，但你的generator1是无限生成器，无法提前生成全部参数，因此不适用你的场景

注意事项 #

1. 共享变量同步：用mp.Value实现多进程间的状态同步，确保找到目标后所有流程能及时终止
2. 任务堆积防控：如果generator1产出num1的速度远快于进程池处理速度，会导致大量待执行任务堆积，可通过调整TASKS_PER_NUM1或给generator1加延迟缓解
3. 终止逻辑选择：pool.terminate()是强制终止，适合你"找到结果立刻停"的需求；如果需要优雅关闭（等待已启动任务完成），可替换为pool.close()，但无法快速终止

内容的提问来源于stack exchange，提问作者MrLatinNerd