蚁群算法是一种启发式优化算法，也是一种智能算法、进化计算。和遗传算法、粒子群算法相比，蚁群算法所优化的内容是拓扑序（或者路径）的信息素浓度，而遗传算法、粒子群算法优化的是某一个个体（解向量）。
例如TSP问题，30个城市之间有900个对应关系，30*15/2=435条路径，在蚂蚁经过之后，都留下了信息素，而某一个拓扑序指的是一个解向量，旅行商的回路应是首尾相连的30条路径，蚂蚁在走路的时候会考虑到能走的路径上的信息素浓度，然后选择一个拓扑序作为新解。例如任务车间调度问题，假如有8个机器和10个任务，则一共有80个对应关系，每个对应关系看作一个路径，会有信息素的标记，而解向量是一个长度为10的向量，蚂蚁在走路的时候，会选择每一个任务到所有机器的对应关系，考虑信息素的浓度选择其中一个，由此选择10个任务各自放在哪一个机器上，作为新解。
蚁群算法天生适合解决路径问题、指派问题，效果通常比粒子群等要好。相比于模拟退火算法，计算更稳定。相比于粒子群算法收敛性更好。
对于任务调度问题，需要定义模型对象：
节点和云任务。节点VM具有节点编号id、两种资源总量和两种资源的已占有量。因此其剩余空间是总量capacity减去已占有量supply。
class Cloudlet: def __init__(self, cpu_demand: float, mem_demand: float): self.cpu_demand = cpu_demand self.mem_demand = mem_demand class VM: def __init__(self, vm_id: int, cpu_supply: float, cpu_velocity: float, mem_supply: float, mem_capacity: float): self.id = vm_id self.cpu_supply = cpu_supply self.cpu_velocity = cpu_velocity self.mem_supply = mem_supply self.mem_capacity = mem_capacity
然后定义蚁群算法解决任务调度问题的类。
初始化属性有云任务列表，节点列表，蚂蚁种群数量，迭代次数，以及人物的路径数和最优策略记录。
函数有生成新解的函数gen_topo_jobs()，评估解的负载均衡值和计算适应度evaluate_particle、calculate_fitness，更新信息素update_topo以及蚁群算法主流程schedulet_main。后面逐一讲解
import numpy as np from typing import List from matplotlib import pyplot as plt class ACScheduler: def __init__(self, cloudlets, vms, population_number=100, times=500): self.cloudlets = cloudlets self.vms = vms self.cloudlet_num = len(cloudlets) # 任务数量也就是粒子长度 self.machine_number = len(vms) # 机器数量 self.population_number = population_number # 种群数量 self.times = times # 迭代代数 # 表示任务选择的机器的信息素 self.topo_phs = [[100 for _ in range(self.machine_number)] for _ in range(self.cloudlet_num)] # 最优策略 self.best_topo = None # 生成新的解向量--根据信息素浓度生成 def gen_topo_jobs(self): pass # 与算法无关的函数--评估当前解向量 def evaluate_particle(self, topo_jobs: List[int]) -> int: pass # 与算法无关的函数--计算适应度 def calculate_fitness(self, topo_jobs: List[int]) -> float: pass # 更新信息素 def update_topo(self): pass def scheduler_main(self): pass if __name__ == '__main__': nodes = [VM(0, 0.862, 950, 950, 1719), VM(1, 0.962, 2, 950, 1719), VM(2, 1.062, 2, 1500, 1719)] lets = [Cloudlet(0.15, 50), Cloudlet(0.05, 100), Cloudlet(0.2, 60), Cloudlet(0.01, 70), Cloudlet(0.04, 80), Cloudlet(0.07, 20), Cloudlet(0.14, 150), Cloudlet(0.15, 200), Cloudlet(0.03, 40), Cloudlet(0.06, 90)] ac = ACScheduler(lets, nodes, times=150) res = ac.scheduler_main() i = 0 for _ in ac.best_topo: print("任务:", i, " 放置到机器", ac.best_topo[i], "上执行") i += 1
1. 生成新的解向量gen_topo_jobs(self)
初始化解向量为-1，然后逐个根据信息素的浓度选择任务的对应的机器，以此生成新的解向量返回
# 生成新的解向量--根据信息素浓度生成 def gen_topo_jobs(self): ans = [-1 for _ in range(self.cloudlet_num)] node_free = [node_id for node_id in range(self.machine_number)] for let in range(self.cloudlet_num): ph_sum = np.sum(list(map(lambda j: self.topo_phs[let][j], node_free))) test_val = 0 rand_ph = np.random.uniform(0, ph_sum) for node_id in node_free: test_val += self.topo_phs[let][node_id] if rand_ph <= test_val: ans[let] = node_id break return ans
2. 评估解向量然后计算适应度
把解向量的策略利用到集群中，然后计算每个节点每个资源的利用率，然后对所有节点取标准差，各个资源的利用率标准差加和作为负载均衡度，越小越均衡。
由于适应度定义为越大越好，因此评估结果取倒数作为适应度。
# 与算法无关的函数--评估当前解向量 def evaluate_particle(self, topo_jobs: List[int]) -> int: cpu_util = np.zeros(self.machine_number) mem_util = np.zeros(self.machine_number) for i in range(len(self.vms)): cpu_util[i] = self.vms[i].cpu_supply mem_util[i] = self.vms[i].mem_supply for i in range(self.cloudlet_num): cpu_util[topo_jobs[i]] += self.cloudlets[i].cpu_demand mem_util[topo_jobs[i]] += self.cloudlets[i].mem_demand for i in range(self.machine_number): if cpu_util[i] > self.vms[i].cpu_velocity: return 100 if mem_util[i] > self.vms[i].mem_capacity: return 100 for i in range(self.machine_number): cpu_util[i] /= self.vms[i].cpu_velocity mem_util[i] /= self.vms[i].mem_capacity return np.std(cpu_util, ddof=1) + np.std(mem_util, ddof=1) # 与算法无关的函数--计算适应度 def calculate_fitness(self, topo_jobs: List[int]) -> float: return 1 / self.evaluate_particle(topo_jobs)
3. 更新信息素：有蚂蚁经过的路径信息素加倍，否则挥发减半
# 更新信息素 def update_topo(self): for i in range(self.cloudlet_num): for j in range(self.machine_number): if j == self.best_topo[i]: self.topo_phs[i][j] *= 2 else: self.topo_phs[i][j] *= 0.5
4. 主流程：迭代求解
def scheduler_main(self): results = [0 for _ in range(self.times)] fitness = 0 for it in range(self.times): best_time = 0 for ant_id in range(self.population_number): topo_jobs = self.gen_topo_jobs() fitness = self.calculate_fitness(topo_jobs) if fitness > best_time: self.best_topo = topo_jobs best_time = fitness assert self.best_topo is not None self.update_topo() results[it] = best_time if it % 10 == 0: print("ACO iter: ", it, " / ", self.times, ", 适应度: ", fitness) plt.plot(range(self.times), results) plt.xlabel("迭代次数") plt.ylabel("适应度") plt.rcParams['font.sans-serif'] = ['KaiTi'] # 指定默认字体 plt.rcParams['axes.unicode_minus'] = False # 解决保存图像是负号'-'显示为方块的问题 plt.title("蚁群算法求解云任务调度负载均衡问题") # plt.savefig('img/ACScheduler-1.1_0.9-popu100-iter200.png', dpi=300, # format='png') # bbox_inches="tight"解决X轴时间两个字不被保存的问题 plt.show() return results
测试数据：
if __name__ == '__main__': # 第四组数据data9 nodes = [ VM(0, 0.662, 2, 620, 2223), VM(1, 0.662, 2, 1100, 2223), VM(2, 1.662, 2, 720, 2223), VM(3, 0.662, 2, 1100, 2223), VM(4, 0.662, 2, 620, 2223), VM(5, 0.562, 2, 650, 2223), VM(6, 0.562, 2, 620, 2223), VM(7, 0.462, 2, 440, 2223), # 8 ] lets = [ Cloudlet(0.133364, 272.435810), Cloudlet(0.226357, 141.126392), Cloudlet(0.084122, 7.183883), Cloudlet(0.029290, 96.658838), Cloudlet(0.027560, 247.821058), Cloudlet(0.191912, 80.636804), Cloudlet(0.134658, 220.702279), Cloudlet(0.133052, 163.046071), Cloudlet(0.272010, 253.477271), Cloudlet(0.175000, 19.409176), Cloudlet(0.166933, 140.880123), Cloudlet(0.286495, 71.288800), Cloudlet(0.080714, 354.839232), Cloudlet(0.209842, 211.351191), Cloudlet(0.221753, 249.500490), Cloudlet(0.128952, 81.599575), Cloudlet(0.168469, 122.216016), Cloudlet(0.049628, 135.728968), Cloudlet(0.051167, 230.172949), Cloudlet(0.158938, 135.356776), Cloudlet(0.212047, 202.830773), Cloudlet(0.372328, 13.145747), Cloudlet(0.092549, 130.122476), Cloudlet(0.166031, 97.761267), Cloudlet(0.142820, 45.852985), Cloudlet(0.016367, 189.495519), Cloudlet(0.112156, 173.926518), Cloudlet(0.004466, 156.806505), Cloudlet(0.222208, 62.619918), Cloudlet(0.073526, 232.175486), Cloudlet(0.158527, 178.624649), Cloudlet(0.103075, 133.896667), Cloudlet(0.176026, 156.076929), Cloudlet(0.098275, 136.450544), Cloudlet(0.192065, 33.923922), Cloudlet(0.213519, 134.820860), ] ac = ACScheduler(lets, nodes, times=150) res = ac.scheduler_main() i = 0 for _ in ac.best_topo: print("任务:", i, " 放置到机器", ac.best_topo[i], "上执行") i += 1
求解结果：
ACO iter: 0 / 150 , 适应度: 0.01
ACO iter: 10 / 150 , 适应度: 4.0408235554621665
ACO iter: 20 / 150 , 适应度: 4.754201711827778
ACO iter: 30 / 150 , 适应度: 4.754201711827778
ACO iter: 40 / 150 , 适应度: 4.754201711827778
ACO iter: 50 / 150 , 适应度: 4.754201711827778
ACO iter: 60 / 150 , 适应度: 4.754201711827778
ACO iter: 70 / 150 , 适应度: 4.754201711827778
ACO iter: 80 / 150 , 适应度: 4.754201711827778
ACO iter: 90 / 150 , 适应度: 4.754201711827778
ACO iter: 100 / 150 , 适应度: 4.754201711827778
ACO iter: 110 / 150 , 适应度: 4.754201711827778
ACO iter: 120 / 150 , 适应度: 4.754201711827778
ACO iter: 130 / 150 , 适应度: 4.754201711827778
ACO iter: 140 / 150 , 适应度: 4.754201711827778
