如何拆分itertools.combinations实现组合结果的并行处理？

我需要处理86选10的海量组合（总计约3.5万亿个结果），已实现每秒处理500,000个组合的算法，但单进程需耗时81天，因此希望利用多核多进程并行处理。

尝试的朴素方案 #

import itertools
from concurrent.futures import ProcessPoolExecutor

def algorithm(combination):
  # returns a boolean in roughly 1/500000th of a second on average

def process(combinations):
  for combination in combinations:
    if algorithm(combination):
      # will be very rare (a few hundred times out of trillions) if that matters
      print("Found matching combination!", combination) 

combination_generator = itertools.combinations(eighty_six_elements, 10)

# My system will have 64 cores and 128 GiB of memory
with ProcessPoolExecutor(workers=63) as executor:
  # assign 1,000,000 combinations to each process
  # it may be more performant to use larger batches (to avoid process startup overhead)
  # but eventually I need to start worrying about running out of memory
  group = []
  for combination in combination_generator:
    group.append(combination)
    if len(group) >= 1_000_000:
      executor.submit(process, group)
      group = []

遇到的瓶颈 #

该代码可运行，但因组合生成环节（for combination in combination_generator）成为瓶颈，几乎无性能提升。

核心问题：
如何将组合生成任务分配给子进程实现并行化？如何让每个进程生成itertools.combinations的指定子集？

附：已找到生成单个指定组合的方案，但需高效生成百万级指定组合。

解决方案：分区间生成组合子集 #

要彻底解决主进程生成组合的瓶颈，核心思路是让每个子进程独立生成自己负责的组合区间——利用组合与唯一序号的映射关系，给每个进程分配一个序号范围，进程直接生成该范围内的所有组合并处理，无需主进程传递组合列表。

关键实现步骤 #

1. 组合与序号的双向映射：实现nth_combination函数，根据序号生成对应的组合（基于你已有的单组合生成方案扩展）
2. 拆分任务区间：计算总组合数，将其拆分为与进程数匹配的子区间，保证每个进程任务量大致均衡
3. 子进程独立处理：每个进程接收自己的起始/结束序号，循环生成组合并调用算法处理

优化后代码示例 #

import math
from concurrent.futures import ProcessPoolExecutor

def algorithm(combination):
    # 你的组合处理逻辑，返回布尔值
    return False

def nth_combination(iterable, r, n):
    # 根据序号n生成对应组合的实现
    pool = tuple(iterable)
    n = int(n)
    k = len(pool)
    if r < 0 or r > k or n < 0 or n >= math.comb(k, r):
        raise ValueError("Invalid parameters for combination generation")
    
    result = []
    remaining = k
    needed = r
    for i in range(k):
        if needed == 0:
            break
        cnt = math.comb(remaining - 1, needed - 1)
        if n < cnt:
            result.append(pool[i])
            needed -= 1
        else:
            n -= cnt
        remaining -= 1
    return tuple(result)

def process_range(start_idx, end_idx, elements, r):
    # 子进程处理指定序号区间的组合
    for idx in range(start_idx, end_idx):
        combo = nth_combination(elements, r, idx)
        if algorithm(combo):
            print("Found matching combination!", combo)

if __name__ == "__main__":
    eighty_six_elements = list(range(86))  # 替换为你的实际元素列表
    r = 10
    total_combinations = math.comb(86, r)
    num_workers = 63
    chunk_size = total_combinations // num_workers
    
    # 拆分任务区间
    tasks = []
    for i in range(num_workers):
        start = i * chunk_size
        # 最后一个进程处理剩余所有组合
        end = start + chunk_size if i < num_workers -1 else total_combinations
        tasks.append( (start, end, eighty_six_elements, r) )
    
    # 启动多进程处理
    with ProcessPoolExecutor(max_workers=num_workers) as executor:
        for task in tasks:
            executor.submit(process_range, *task)

核心优化点 #

• 消除主进程瓶颈：每个子进程独立生成组合，主进程仅负责拆分区间，不再承担组合生成的计算压力
• 内存高效：无需在主进程缓存大量组合，子进程按需生成单个组合并处理，内存占用极低
• 负载均衡：通过均分序号区间，保证每个进程的任务量大致相等，充分利用64核资源

额外建议 #

• 若需要持久化结果，建议使用多进程安全的队列或带锁的文件写入，避免print导致的输出混乱
• 可以在process_range内增加小批次处理逻辑（比如每处理100万组合做一次状态检查），方便监控任务进度

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