如何并行化以指定结果数量为终止条件的未知调用次数While循环？

我写了一个执行计算的函数，每次调用时因为使用不同的随机数生成器（rng）种子，返回结果都不一样。通常我需要多次调用它来获取大量样本。

我已经用multiprocessing模块实现了这个函数的并行调用，比如用4个进程，直到达到预设的调用次数n_runs。下面是最小可运行示例（注：flip_coin()只是一个用了rng的示例函数，实际用的是更复杂的函数）：

import multiprocessing as mp
import random, sys

def flip_coin(n):    
    # 初始化随机数生成器
    seed = random.randrange(sys.maxsize)
    rng = random.Random(seed)
    # 执行操作并获取结果
    if rng.random()>0.5: res = 1
    else: res = 0
    return res, seed

# 总调用次数
n_runs = 100
# 初始化进程池
pool = mp.Pool(processes = 4)
# 初始化结果存储列表
results, seeds = [], []
for result in pool.map(flip_coin, range(n_runs)):
    # 保存结果及生成该结果的种子
    results.append(result[0])
    seeds.append(result[1])
# 关闭进程池
pool.close(); pool.join() 

现在我不想预先固定n_runs，而是想用另一种条件终止循环，这个条件需要调用函数未知次数后才能满足。比如，我希望收集函数返回的指定数量的1。不用multiprocessing的话，我会这么写：

# 目标1的数量
n_ones = 10
# 统计1的计数器
counter = 0
# 存储种子的空列表
seeds = []
while counter < n_ones:
    result = flip_coin(1)
    # 若得到1，则增加计数器并保存种子
    if result[0] == 1: 
        counter += 1
        seeds.append(result[1])

请问怎么把这类While循环并行化？

解决方案 #

要实现这种基于动态条件的并行循环，核心是实时获取进程返回的结果并判断终止条件，而非等待所有预设任务完成。可以借助multiprocessing.Pool的imap_unordered方法，它能按任务完成顺序返回结果，让我们在收集到足够目标结果后立即停止所有进程。

实现代码 #

import multiprocessing as mp
import random, sys

def flip_coin(_):    
    # 初始化随机数生成器
    seed = random.randrange(sys.maxsize)
    rng = random.Random(seed)
    res = 1 if rng.random() > 0.5 else 0
    return res, seed

def main():
    # 目标1的数量
    target_ones = 10
    collected_ones = 0
    seeds = []
    
    # 用with语句管理进程池，自动处理关闭/回收
    with mp.Pool(processes=4) as pool:
        # 生成无限任务迭代器：给函数传占位参数，持续生成任务直到主动终止
        tasks = iter(lambda: 1, None)
        # 实时遍历完成的任务结果
        for res, seed in pool.imap_unordered(flip_coin, tasks):
            if res == 1:
                collected_ones += 1
                seeds.append(seed)
                # 满足条件立即终止所有进程，停止剩余任务
                if collected_ones >= target_ones:
                    pool.terminate()
                    break
    
    print(f"收集到{target_ones}个1，对应的种子：{seeds}")

if __name__ == "__main__":
    main()

关键细节 #

• imap_unordered：区别于map的批量返回，它会在单个任务完成后立刻返回结果，这是实现动态终止的核心。
• pool.terminate()：满足终止条件时调用，可立即终止所有子进程，避免无意义的计算开销。
• 无限任务迭代器：用iter(lambda: 1, None)生成持续输出占位参数的迭代器，让进程池不断生成任务，直到我们主动中断。

优化提示 #

如果你的实际函数执行耗时较长，可以改为批量生成任务（比如每次生成20个），减少进程池的调度开销，核心逻辑仍保持实时判断结果并终止的流程。

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