Python如何用多进程加速超大permutations迭代器的过滤计数任务

多进程/多线程选型说明 #

• 你的任务属于CPU密集型计算，Python多线程受GIL（全局解释器锁）限制，无法真正利用多核CPU，所以多进程是唯一可选的并行方案，你之前的选型方向是对的，问题出在使用逻辑不合理。

原有多进程方案性能差的核心原因 #

你直接把排列迭代器的单个元素提交给子进程，存在两个致命问题：

1. 单个任务计算量极小：f函数的执行耗时远低于进程间数据传输（IPC）的开销，并行收益完全被通信成本覆盖，所以速度反而比单进程更慢。
2. pool.map()会默认将整个迭代器转为列表再拆分提交，超大的排列迭代器直接被全量加载到内存，导致内存占满。

优化实现方案 #

核心优化逻辑是批量提交任务，每次给子进程传入一批排列，大幅降低IPC的次数，同时避免全量加载迭代器。

第一步：优化判断函数（可选但收益明显） #

原有join成字符串再判断的写法可以进一步优化，减少不必要的字符串生成开销：

def has_ab(perm):
    for i in range(len(perm)-1):
        if perm[i] == 'A' and perm[i+1] == 'B':
            return True
    return False

第二步：实现批量处理逻辑 #

from itertools import permutations, islice
from multiprocessing import Pool, cpu_count

# 批量判断函数，子进程每次处理一批排列
def batch_worker(batch):
    count = 0
    for perm in batch:
        for i in range(len(perm)-1):
            if perm[i] == 'A' and perm[i+1] == 'B':
                count +=1
                break
    return count

# 迭代器分块生成器，不加载全量数据，每次返回一个批次的排列
def chunk_iterator(iterator, chunk_size=10000):
    while True:
        chunk = list(islice(iterator, chunk_size))
        if not chunk:
            break
        yield chunk

if __name__ == "__main__":
    # 替换为你的原始排列迭代器
    iterator = permutations(['A', 'B', 'C', 'D', 'E', 'F', 'G', 'H'])
    # 块大小可根据实际情况调整，一般1000~100000区间测试最优值
    chunk_size = 20000
    # 进程数设置为CPU核心数即可
    pool = Pool(cpu_count())
    total = 0
    # 用imap_unordered异步获取结果，不要求顺序进一步提升速度
    for batch_count in pool.imap_unordered(batch_worker, chunk_iterator(iterator, chunk_size)):
        total += batch_count
    pool.close()
    pool.join()
    print(total)

参数调整建议 #

• chunk_size是核心调优参数：如果块太小，通信开销还是高；如果块太大，可能出现部分进程早早跑完闲置、负载不均衡的问题，建议在1000~100000的区间内测试，找到当前场景下的最优值。
• 进程数默认用cpu_count()即可，不需要额外调大，超过核心数会导致进程切换开销升高。

额外优化建议（如果你的判断逻辑是固定的相邻字符匹配） #

如果你的需求只是统计排列中存在连续AB子序列的数量，完全可以通过组合数学公式直接计算，不需要遍历所有排列：
假设你有n个不重复的元素，其中恰好包含1个A和1个B，那么包含连续AB的排列总数为 (n-1)!，直接计算阶乘即可得到结果，效率远高于任何遍历+并行方案。

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