简介一套基于Python实现的并行遗传算法资料适合算法爱好者、Python开发者以及需要求解大规模优化问题的研究人员。资源覆盖遗传算法核心流程——种群初始化、适应度评价、选择、交叉与变异并演示借助multiprocessing或concurrent.futures将种群分割到多进程/线程中并行计算以降低耗时、提升收敛效率。压缩包共2个文件均为py脚本体积仅6KB代码量精简、易于阅读可作为理解PGA并行思路与快速改造的基础模板。资源目前已有224人学习浏览适合用于课程设计、毕业设计或工程预研阶段的算法验证。通过阅读源码可重点体会任务划分、负载均衡、数据同步以及并行效率评估等关键细节避免常见并行陷阱。1. 并行遗传算法为什么是Python性能问题的实际解法遗传算法GA是处理非线性、多峰、离散组合优化问题的常用工具但跑起来的等待时间经常让人抓狂种群里几百个个体进化几十代就要评估几千到几万次适应度。如果你的评估函数是一次仿真或一轮机器学习训练总时长从分钟变成小时是常事。在Python里这种等待还被解释器开销放大最气人的是服务器的多核CPU大部分时间只有一核在工作。并行遗传算法Parallel Genetic AlgorithmPGA把每个个体的适应度评估分散到多个进程同时完成在单机多核上就能拿到接近线性的加速比也是后续做分布式计算的自然过渡。这篇内容适合已经写过基础GA、想用Python优化耗时代码或者正准备在DEAP等框架里引入并行的工程师。我们不推倒重来而是在已能运行的GA代码上做最小改造。2. 先打下基线用Python实现标准遗传算法并行化不是让一个有bug的算法跑得更快而是让正确算法的扩展性变好。所以在讨论多进程之前先有一个可验证的串行实现作为基准。这一节我们用30维Rastrigin函数作为测试目标它的全局最小值在原点周围有大量局部极小值是评估GA收敛能力的常用压力测试。2.1 个体编码与适应度函数设计Rastrigin函数定义为f(x) A*n Σ(x_i^2 - A*cos(2πx_i))其中取 A10解空间为 [-5.12, 5.12]^n。我们对每个个体用Python列表存浮点值适应度返回函数值本身因为要求最小值。代码如下import random import time import numpy as np def eval_rastrigin(individual): A 10 n len(individual) val A * n sum(xi * xi - A * np.cos(2 * np.pi * xi) for xi in individual) return (val,) # DEAP要求返回tuple这里的(val,)是DEAP中适应度统一用元组表示后续weights为(-1.0,)就代表是求最小化。适应度函数越慢并行化的收益越明显。如果你现有的评估函数是仿真或SQL查询这一步通常已经是瓶颈了。2.2 选择、交叉、变异的串行实现通常我们不自己重写遗传算子用DEAP提供的高质量实现。先注册几个算子from deap import base, creator, tools, algorithms creator.create(FitnessMin, base.Fitness, weights(-1.0,)) creator.create(Individual, list, fitnesscreator.FitnessMin) toolbox base.Toolbox() toolbox.register(attr_float, random.uniform, -5.12, 5.12) toolbox.register(individual, tools.initRepeat, creator.Individual, toolbox.attr_float, n30) toolbox.register(population, tools.initRepeat, list, toolbox.individual) toolbox.register(evaluate, eval_rastrigin) # 模拟二进制交叉SBX适合实数编码 toolbox.register(mate, tools.cxSimulatedBinaryBounded, low-5.12, up5.12, eta20.0) # 高斯变异sigma1.0表示扰动幅度indpb控制每个基因位变异概率 toolbox.register(mutate, tools.mutGaussian, mu0, sigma1.0, indpb0.2) # 锦标赛选择tournsize3 toolbox.register(select, tools.selTournament, tournsize3)各参数的含义eta20.0是SBX的分布指数值越大子代越接近父代indpb0.2意味着30个基因中平均6个被扰动锦标赛选择size3是经典配置。这些参数对解的质量影响很大但对于并行改造本身保持默认即可。2.3 串行主循环与基准测试下面跑30代种群规模100交叉概率0.6变异概率0.3并打印每代最小值和总耗时def run_serial(): pop toolbox.population(n100) hof tools.HallOfFame(1) stats tools.Statistics(lambda ind: ind.fitness.values) stats.register(min, np.min) stats.register(avg, np.mean) start time.time() pop, log algorithms.eaSimple(pop, toolbox, cxpb0.6, mutpb0.3, ngen30, statsstats, halloffamehof, verboseTrue) elapsed time.time() - start print(fSerial time: {elapsed:.2f} s) return hof[0].fitness.values[0]eaSimple是DEAP最基础的进化算法流程先用select选父本再做交叉和变异最后用HallOfFame保留历史最优个体。这个流程没有任何分布式逻辑是我们做并行改造的参照物。把这段代码保存为ga_serial.py记下执行时间。如果这里就跑不动先别做并行去优化适应度函数本身比如用pandas切片代替纯Python循环或把重复计算移出函数。串行运行时的资源特征用下面这个表格来总结指标串行运行值CPU使用率约1核100%单次适应度评估假设0.01s总评估次数100个体 * 30代 3000总耗时约30s 选择交叉时间3. 并行遗传算法GA的四种模型与Python选型串行GA循环里最清晰的加速点就是适应度评估一个generation里100个个体互相独立计算完才进入下一代。并行GA通常分成四大类它们在Python社区都有对应实现但复杂度差异很大。3.1 主从模型主进程负责选择、交叉、变异和种群管理把当前种群发给多个从进程并行计算适应度。这个模型对多核机器最友好也最容易改造在上节代码中只用把toolbox.register(map, pool.map)其余完全不动。主从模型的问题是同步开销每一代都需要收集全部结果主进程等待最慢的那个个体。当评估时间在0.001秒以下、通信和进程切换占主导时加速比反而下降。3.2 岛屿模型把整个种群分成若干子种群各自独立在单独进程/线程上跑完整代的GA每隔几代把“最佳个体”迁移到邻居岛屿替换对方种群中的较差个体。迁移率、迁移拓扑和间隔是新增的三个参数。岛屿模型保留了群体多样性不容易早熟但每个岛需要稳定运行进程间的通信频率远低于主从模型适合评估时间短但种群数量大的情况。岛屿模型的伪代码框架通常是这样的# 岛屿模型的伪代码框架 def island_evolve(island_id): pop init_population(island_id) for gen in range(NGEN): pop evolve_one_generation(pop) if gen % MIGRATION_INTERVAL 0: send_best_to_neighbor(island_id, pop) recv_and_replace(pop, island_id)这里MIGRATION_INTERVAL控制迁移频率send_best_to_neighbor负责把当前岛最优个体发给相邻岛recv_and_replace用收到的个体替换最差个体。实际实现时可以用multiprocessing.Queue或消息中间件但单机场景下用Python的mp.Pipe更容易控制。3.3 细胞模型个体分布在二维网格上每个个体只与邻近的少数个体做选择和交叉本质上是细粒度的并行。这种模型在研究GA的种群结构时很有用但在工程实践中很少单独实现因为Python操作二维邻域的开销比数值计算还大。如果你的场景是GPU上大规模并行细胞模型才有实现价值。3.4 怎么用CPU拓扑决定选型工程选型我有一条经验评估耗时大于10ms时优先主从评估很快但收敛不稳定用岛屿手头只有一个带GPU的机器且个体数上万再考虑细胞。这里还要区分“并行”和“并发”Python的threading受GIL限制不适合跑纯Python密集的适应度评估所以不要浪费时间改线程版本直接用多进程。各模型特点对比如下模型适用范围Python实现成本风险主从适应度耗时长、个体间完全独立低仅注册map同步等待、进程间传输大对象岛屿种群多样性强、子代质量要求高中需维护多个种群迁移参数难调细胞计算资源多、需要细粒度控制高不建议单机邻域更新开销大4. 用multiprocessing和DEAP实现主从并行GA这一节直接给出可运行的完整代码在类似上一章的串行版本上改成并行版本并带计时和结果输出。4.1 安装与导入如果环境里还没有DEAP先安装pip install deapDEAP版本目前官方稳定在1.4不需要额外装并行库Python标准库的multiprocessing够用。如果是macOS或Linux程序入口需要这样写import multiprocessing as mp import time import random import numpy as np from deap import base, creator, tools, algorithms # 必须放在模块顶层因为multiprocessing需要pickle子进程的函数 def eval_rastrigin(individual): time.sleep(0.01) # 模拟真实仿真耗时10ms A 10 n len(individual) val A * n sum(xi * xi - A * np.cos(2 * np.pi * xi) for xi in individual) return (val,)这里我在适应度函数里加了一行sleep(0.01)来模拟仿真或训练类的评估实际项目中它可能是一个需要跑100步的模拟器也可能是一个需要推理几百张图片的神经网络。这个10ms对应3000次评估就是30秒足够看出并行效果。4.2 注册算子与并行map遗传算子跟串行完全相同creator.create(FitnessMin, base.Fitness, weights(-1.0,)) creator.create(Individual, list, fitnesscreator.FitnessMin) toolbox base.Toolbox() toolbox.register(attr_float, random.uniform, -5.12, 5.12) toolbox.register(individual, tools.initRepeat, creator.Individual, toolbox.attr_float, n30) toolbox.register(population, tools.initRepeat, list, toolbox.individual) toolbox.register(evaluate, eval_rastrigin) toolbox.register(mate, tools.cxSimulatedBinaryBounded, low-5.12, up5.12, eta20.0) toolbox.register(mutate, tools.mutGaussian, mu0, sigma1.0, indpb0.2) toolbox.register(select, tools.selTournament, tournsize3)关键改变在主函数里def main(): pop toolbox.population(n100) hof tools.HallOfFame(1) stats tools.Statistics(lambda ind: ind.fitness.values) stats.register(min, np.min) stats.register(avg, np.mean) # 创建进程池进程数按CPU核数-1留一个给系统 pool mp.Pool(processesmp.cpu_count() - 1) toolbox.register(map, pool.map) start time.time() pop, log algorithms.eaSimple(pop, toolbox, cxpb0.6, mutpb0.3, ngen30, statsstats, halloffamehof, verboseTrue) elapsed time.time() - start print(fParallel time: {elapsed:.2f} s) print(fBest fitness: {hof[0].fitness.values[0]:.6f}) pool.close() pool.join() if __name__ __main__: main()mp.Pool(processes...)创建了工作进程池。toolbox.register(map, pool.map)把DEAP内部原来用内建map逐个调用evaluate的地方替换成进程池的分布map。当eaSimple对每个个体调用适应度函数时实际上是池分配任务给子进程并行执行。注意mp.cpu_count() - 1只是通用选择如果评估函数本身是纯Python循环受内存带宽限制4个进程可能就饱和了。4.3 同步与异步评估的取舍pool.map是同步阻塞的每代必须等所有个体都评估完主进程才继续。这正好满足标准GA的同步进化逻辑。如果适应度评估时间不均衡——比如实际仿真中有些算例需要更多步数收敛——可以用imap_unordered并自己实现类似eaMuPlusLambda的流程。但DEAP内建算法接受map的返回值必须是按输入顺序对齐的列表所以异步评估需要重写循环。我的建议是先保持同步等确认正确再优化。4.4 关键参数进程数、chunksize和内存进程数并不是越多越好。mp.cpu_count() - 1只是通用选择如果评估函数本身是纯Python循环受内存带宽限制4个进程可能就饱和了。可以在pool.map的调用中指定chunksize10让池一次给子进程分发10个个体减少IPC次数。但由于toolbox.register(map, pool.map)没法直接传chunksize建议用functools.partialfrom functools import partial toolbox.register(map, partial(pool.map, chunksize10))另一种做法是使用imap_unordered并自己包装成list但保持顺序但最简单的是调整种群大小到进程数的整数倍。运行完上面的代码你会看到每代耗时的变化。下面是一个典型8核笔记本上的对比评估时间为模拟的10ms仅供参考运行版本每代耗时(秒)30代总耗时(秒)加速比串行1.0301x并行4进程0.288.43.6x并行7进程0.185.45.5x为什么不是7倍因为GA中除了评估还有选择和交叉的串行部分受Amdahl定律约束进程启动、数据序列化和合并结果也有开销。如果评估从10ms改为100ms加速比会更接近进程数。5. 在项目里落地调参、验证与常遇到的坑最后要提几个真正跑项目时会踩的坑。有些是我自己踩过的写出来省得你再趟。5.1 主进程是否参与计算mp.Pool创建的子进程是独立进程主进程只是调度不参与评估。如果你的机器有8核用processes8时主进程也是闲置的长时间开着浪费一块CPU。通常选择mp.cpu_count()-1让主进程也处理少量任务但DEAP的简单模型做不到主进程承担评估。要榨干最后一核可以手动把种群一分为二一部分用本进程评估一部分丢给pool但这会让代码复杂且收益仅多1/8。5.2 随机数可复现性multiprocessing的子进程各自继承父进程的随机状态导致每次运行结果不一致。要复现实验在main()开头设置固定种子random.seed(42) np.random.seed(42)但这只能保证基因初始化可控子进程里的time.sleep模拟不会影响结果。如果适应度函数内部使用random还需要给每个子进程单独设置种子常用做法是让每个worker用固定的偏移量例如def init_worker(worker_id): random.seed(42 worker_id)并在Pool(initializerinit_worker, initargs(i,))中声明。5.3 进程间传输大对象如果individual不是30个float而是包含几千个坐标点的复杂对象pool.map每次传输都会pickle整个个体内存和延迟都会升高。可以尝试把个体转成numpy数组并用array协议传输或者把评估函数设计成接收索引从共享内存读取数据。multiprocessing.Array和shared_memory可以做到零拷贝但对编码要求高。5.4 一个高级技巧用缓存避免重复评估GA在收敛后期同一适应度的个体会反复出现。在适应度函数外面加一层lru_cache用个体元组作为key可以明显减少计算量。不过Python的functools.lru_cache只对哈希参数生效list不行先把individual转成tuple再存。from functools import lru_cache lru_cache(maxsize10000) def cached_eval(individual_tuple): # 调用真正的评估逻辑 return eval_rastrigin(list(individual_tuple)) # 在注册evaluate时换成缓存包装 def eval_with_cache(individual): return cached_eval(tuple(individual)) toolbox.register(evaluate, eval_with_cache)注意lru_cache是线程安全的但在多进程下各进程有自己的缓存不会共享因此只在串行或单worker场景下能发挥全部效果。如果做分布式并行需要在进程间共享缓存复杂度就高了通常直接用Redis整一个LRU代替。验证并行GA是否有效不要看单次运行时间要看多次运行的平均时间与方差。并行引入的随机性会让运行时间抖动尤其是操作系统调度和进程竞争。最后留一个检查清单一确认结果与串行版的适应度曲线保持同一量级二用不同进程数跑三遍取平均三把verboseTrue打出来的每秒评估数相比串行是否有提升四任务管理器观察CPU占用是否接近目标百分比。只有这些都对上了才算真正落了一个并行GA。本文还有配套的精品资源点击获取 Python并行遗传算法实战：多进程加速进化计算的全流程指南
简介一套基于Python实现的并行遗传算法资料适合算法爱好者、Python开发者以及需要求解大规模优化问题的研究人员。资源覆盖遗传算法核心流程——种群初始化、适应度评价、选择、交叉与变异并演示借助multiprocessing或concurrent.futures将种群分割到多进程/线程中并行计算以降低耗时、提升收敛效率。压缩包共2个文件均为py脚本体积仅6KB代码量精简、易于阅读可作为理解PGA并行思路与快速改造的基础模板。资源目前已有224人学习浏览适合用于课程设计、毕业设计或工程预研阶段的算法验证。通过阅读源码可重点体会任务划分、负载均衡、数据同步以及并行效率评估等关键细节避免常见并行陷阱。1. 并行遗传算法为什么是Python性能问题的实际解法遗传算法GA是处理非线性、多峰、离散组合优化问题的常用工具但跑起来的等待时间经常让人抓狂种群里几百个个体进化几十代就要评估几千到几万次适应度。如果你的评估函数是一次仿真或一轮机器学习训练总时长从分钟变成小时是常事。在Python里这种等待还被解释器开销放大最气人的是服务器的多核CPU大部分时间只有一核在工作。并行遗传算法Parallel Genetic AlgorithmPGA把每个个体的适应度评估分散到多个进程同时完成在单机多核上就能拿到接近线性的加速比也是后续做分布式计算的自然过渡。这篇内容适合已经写过基础GA、想用Python优化耗时代码或者正准备在DEAP等框架里引入并行的工程师。我们不推倒重来而是在已能运行的GA代码上做最小改造。2. 先打下基线用Python实现标准遗传算法并行化不是让一个有bug的算法跑得更快而是让正确算法的扩展性变好。所以在讨论多进程之前先有一个可验证的串行实现作为基准。这一节我们用30维Rastrigin函数作为测试目标它的全局最小值在原点周围有大量局部极小值是评估GA收敛能力的常用压力测试。2.1 个体编码与适应度函数设计Rastrigin函数定义为f(x) A*n Σ(x_i^2 - A*cos(2πx_i))其中取 A10解空间为 [-5.12, 5.12]^n。我们对每个个体用Python列表存浮点值适应度返回函数值本身因为要求最小值。代码如下import random import time import numpy as np def eval_rastrigin(individual): A 10 n len(individual) val A * n sum(xi * xi - A * np.cos(2 * np.pi * xi) for xi in individual) return (val,) # DEAP要求返回tuple这里的(val,)是DEAP中适应度统一用元组表示后续weights为(-1.0,)就代表是求最小化。适应度函数越慢并行化的收益越明显。如果你现有的评估函数是仿真或SQL查询这一步通常已经是瓶颈了。2.2 选择、交叉、变异的串行实现通常我们不自己重写遗传算子用DEAP提供的高质量实现。先注册几个算子from deap import base, creator, tools, algorithms creator.create(FitnessMin, base.Fitness, weights(-1.0,)) creator.create(Individual, list, fitnesscreator.FitnessMin) toolbox base.Toolbox() toolbox.register(attr_float, random.uniform, -5.12, 5.12) toolbox.register(individual, tools.initRepeat, creator.Individual, toolbox.attr_float, n30) toolbox.register(population, tools.initRepeat, list, toolbox.individual) toolbox.register(evaluate, eval_rastrigin) # 模拟二进制交叉SBX适合实数编码 toolbox.register(mate, tools.cxSimulatedBinaryBounded, low-5.12, up5.12, eta20.0) # 高斯变异sigma1.0表示扰动幅度indpb控制每个基因位变异概率 toolbox.register(mutate, tools.mutGaussian, mu0, sigma1.0, indpb0.2) # 锦标赛选择tournsize3 toolbox.register(select, tools.selTournament, tournsize3)各参数的含义eta20.0是SBX的分布指数值越大子代越接近父代indpb0.2意味着30个基因中平均6个被扰动锦标赛选择size3是经典配置。这些参数对解的质量影响很大但对于并行改造本身保持默认即可。2.3 串行主循环与基准测试下面跑30代种群规模100交叉概率0.6变异概率0.3并打印每代最小值和总耗时def run_serial(): pop toolbox.population(n100) hof tools.HallOfFame(1) stats tools.Statistics(lambda ind: ind.fitness.values) stats.register(min, np.min) stats.register(avg, np.mean) start time.time() pop, log algorithms.eaSimple(pop, toolbox, cxpb0.6, mutpb0.3, ngen30, statsstats, halloffamehof, verboseTrue) elapsed time.time() - start print(fSerial time: {elapsed:.2f} s) return hof[0].fitness.values[0]eaSimple是DEAP最基础的进化算法流程先用select选父本再做交叉和变异最后用HallOfFame保留历史最优个体。这个流程没有任何分布式逻辑是我们做并行改造的参照物。把这段代码保存为ga_serial.py记下执行时间。如果这里就跑不动先别做并行去优化适应度函数本身比如用pandas切片代替纯Python循环或把重复计算移出函数。串行运行时的资源特征用下面这个表格来总结指标串行运行值CPU使用率约1核100%单次适应度评估假设0.01s总评估次数100个体 * 30代 3000总耗时约30s 选择交叉时间3. 并行遗传算法GA的四种模型与Python选型串行GA循环里最清晰的加速点就是适应度评估一个generation里100个个体互相独立计算完才进入下一代。并行GA通常分成四大类它们在Python社区都有对应实现但复杂度差异很大。3.1 主从模型主进程负责选择、交叉、变异和种群管理把当前种群发给多个从进程并行计算适应度。这个模型对多核机器最友好也最容易改造在上节代码中只用把toolbox.register(map, pool.map)其余完全不动。主从模型的问题是同步开销每一代都需要收集全部结果主进程等待最慢的那个个体。当评估时间在0.001秒以下、通信和进程切换占主导时加速比反而下降。3.2 岛屿模型把整个种群分成若干子种群各自独立在单独进程/线程上跑完整代的GA每隔几代把“最佳个体”迁移到邻居岛屿替换对方种群中的较差个体。迁移率、迁移拓扑和间隔是新增的三个参数。岛屿模型保留了群体多样性不容易早熟但每个岛需要稳定运行进程间的通信频率远低于主从模型适合评估时间短但种群数量大的情况。岛屿模型的伪代码框架通常是这样的# 岛屿模型的伪代码框架 def island_evolve(island_id): pop init_population(island_id) for gen in range(NGEN): pop evolve_one_generation(pop) if gen % MIGRATION_INTERVAL 0: send_best_to_neighbor(island_id, pop) recv_and_replace(pop, island_id)这里MIGRATION_INTERVAL控制迁移频率send_best_to_neighbor负责把当前岛最优个体发给相邻岛recv_and_replace用收到的个体替换最差个体。实际实现时可以用multiprocessing.Queue或消息中间件但单机场景下用Python的mp.Pipe更容易控制。3.3 细胞模型个体分布在二维网格上每个个体只与邻近的少数个体做选择和交叉本质上是细粒度的并行。这种模型在研究GA的种群结构时很有用但在工程实践中很少单独实现因为Python操作二维邻域的开销比数值计算还大。如果你的场景是GPU上大规模并行细胞模型才有实现价值。3.4 怎么用CPU拓扑决定选型工程选型我有一条经验评估耗时大于10ms时优先主从评估很快但收敛不稳定用岛屿手头只有一个带GPU的机器且个体数上万再考虑细胞。这里还要区分“并行”和“并发”Python的threading受GIL限制不适合跑纯Python密集的适应度评估所以不要浪费时间改线程版本直接用多进程。各模型特点对比如下模型适用范围Python实现成本风险主从适应度耗时长、个体间完全独立低仅注册map同步等待、进程间传输大对象岛屿种群多样性强、子代质量要求高中需维护多个种群迁移参数难调细胞计算资源多、需要细粒度控制高不建议单机邻域更新开销大4. 用multiprocessing和DEAP实现主从并行GA这一节直接给出可运行的完整代码在类似上一章的串行版本上改成并行版本并带计时和结果输出。4.1 安装与导入如果环境里还没有DEAP先安装pip install deapDEAP版本目前官方稳定在1.4不需要额外装并行库Python标准库的multiprocessing够用。如果是macOS或Linux程序入口需要这样写import multiprocessing as mp import time import random import numpy as np from deap import base, creator, tools, algorithms # 必须放在模块顶层因为multiprocessing需要pickle子进程的函数 def eval_rastrigin(individual): time.sleep(0.01) # 模拟真实仿真耗时10ms A 10 n len(individual) val A * n sum(xi * xi - A * np.cos(2 * np.pi * xi) for xi in individual) return (val,)这里我在适应度函数里加了一行sleep(0.01)来模拟仿真或训练类的评估实际项目中它可能是一个需要跑100步的模拟器也可能是一个需要推理几百张图片的神经网络。这个10ms对应3000次评估就是30秒足够看出并行效果。4.2 注册算子与并行map遗传算子跟串行完全相同creator.create(FitnessMin, base.Fitness, weights(-1.0,)) creator.create(Individual, list, fitnesscreator.FitnessMin) toolbox base.Toolbox() toolbox.register(attr_float, random.uniform, -5.12, 5.12) toolbox.register(individual, tools.initRepeat, creator.Individual, toolbox.attr_float, n30) toolbox.register(population, tools.initRepeat, list, toolbox.individual) toolbox.register(evaluate, eval_rastrigin) toolbox.register(mate, tools.cxSimulatedBinaryBounded, low-5.12, up5.12, eta20.0) toolbox.register(mutate, tools.mutGaussian, mu0, sigma1.0, indpb0.2) toolbox.register(select, tools.selTournament, tournsize3)关键改变在主函数里def main(): pop toolbox.population(n100) hof tools.HallOfFame(1) stats tools.Statistics(lambda ind: ind.fitness.values) stats.register(min, np.min) stats.register(avg, np.mean) # 创建进程池进程数按CPU核数-1留一个给系统 pool mp.Pool(processesmp.cpu_count() - 1) toolbox.register(map, pool.map) start time.time() pop, log algorithms.eaSimple(pop, toolbox, cxpb0.6, mutpb0.3, ngen30, statsstats, halloffamehof, verboseTrue) elapsed time.time() - start print(fParallel time: {elapsed:.2f} s) print(fBest fitness: {hof[0].fitness.values[0]:.6f}) pool.close() pool.join() if __name__ __main__: main()mp.Pool(processes...)创建了工作进程池。toolbox.register(map, pool.map)把DEAP内部原来用内建map逐个调用evaluate的地方替换成进程池的分布map。当eaSimple对每个个体调用适应度函数时实际上是池分配任务给子进程并行执行。注意mp.cpu_count() - 1只是通用选择如果评估函数本身是纯Python循环受内存带宽限制4个进程可能就饱和了。4.3 同步与异步评估的取舍pool.map是同步阻塞的每代必须等所有个体都评估完主进程才继续。这正好满足标准GA的同步进化逻辑。如果适应度评估时间不均衡——比如实际仿真中有些算例需要更多步数收敛——可以用imap_unordered并自己实现类似eaMuPlusLambda的流程。但DEAP内建算法接受map的返回值必须是按输入顺序对齐的列表所以异步评估需要重写循环。我的建议是先保持同步等确认正确再优化。4.4 关键参数进程数、chunksize和内存进程数并不是越多越好。mp.cpu_count() - 1只是通用选择如果评估函数本身是纯Python循环受内存带宽限制4个进程可能就饱和了。可以在pool.map的调用中指定chunksize10让池一次给子进程分发10个个体减少IPC次数。但由于toolbox.register(map, pool.map)没法直接传chunksize建议用functools.partialfrom functools import partial toolbox.register(map, partial(pool.map, chunksize10))另一种做法是使用imap_unordered并自己包装成list但保持顺序但最简单的是调整种群大小到进程数的整数倍。运行完上面的代码你会看到每代耗时的变化。下面是一个典型8核笔记本上的对比评估时间为模拟的10ms仅供参考运行版本每代耗时(秒)30代总耗时(秒)加速比串行1.0301x并行4进程0.288.43.6x并行7进程0.185.45.5x为什么不是7倍因为GA中除了评估还有选择和交叉的串行部分受Amdahl定律约束进程启动、数据序列化和合并结果也有开销。如果评估从10ms改为100ms加速比会更接近进程数。5. 在项目里落地调参、验证与常遇到的坑最后要提几个真正跑项目时会踩的坑。有些是我自己踩过的写出来省得你再趟。5.1 主进程是否参与计算mp.Pool创建的子进程是独立进程主进程只是调度不参与评估。如果你的机器有8核用processes8时主进程也是闲置的长时间开着浪费一块CPU。通常选择mp.cpu_count()-1让主进程也处理少量任务但DEAP的简单模型做不到主进程承担评估。要榨干最后一核可以手动把种群一分为二一部分用本进程评估一部分丢给pool但这会让代码复杂且收益仅多1/8。5.2 随机数可复现性multiprocessing的子进程各自继承父进程的随机状态导致每次运行结果不一致。要复现实验在main()开头设置固定种子random.seed(42) np.random.seed(42)但这只能保证基因初始化可控子进程里的time.sleep模拟不会影响结果。如果适应度函数内部使用random还需要给每个子进程单独设置种子常用做法是让每个worker用固定的偏移量例如def init_worker(worker_id): random.seed(42 worker_id)并在Pool(initializerinit_worker, initargs(i,))中声明。5.3 进程间传输大对象如果individual不是30个float而是包含几千个坐标点的复杂对象pool.map每次传输都会pickle整个个体内存和延迟都会升高。可以尝试把个体转成numpy数组并用array协议传输或者把评估函数设计成接收索引从共享内存读取数据。multiprocessing.Array和shared_memory可以做到零拷贝但对编码要求高。5.4 一个高级技巧用缓存避免重复评估GA在收敛后期同一适应度的个体会反复出现。在适应度函数外面加一层lru_cache用个体元组作为key可以明显减少计算量。不过Python的functools.lru_cache只对哈希参数生效list不行先把individual转成tuple再存。from functools import lru_cache lru_cache(maxsize10000) def cached_eval(individual_tuple): # 调用真正的评估逻辑 return eval_rastrigin(list(individual_tuple)) # 在注册evaluate时换成缓存包装 def eval_with_cache(individual): return cached_eval(tuple(individual)) toolbox.register(evaluate, eval_with_cache)注意lru_cache是线程安全的但在多进程下各进程有自己的缓存不会共享因此只在串行或单worker场景下能发挥全部效果。如果做分布式并行需要在进程间共享缓存复杂度就高了通常直接用Redis整一个LRU代替。验证并行GA是否有效不要看单次运行时间要看多次运行的平均时间与方差。并行引入的随机性会让运行时间抖动尤其是操作系统调度和进程竞争。最后留一个检查清单一确认结果与串行版的适应度曲线保持同一量级二用不同进程数跑三遍取平均三把verboseTrue打出来的每秒评估数相比串行是否有提升四任务管理器观察CPU占用是否接近目标百分比。只有这些都对上了才算真正落了一个并行GA。本文还有配套的精品资源点击获取