如何在OptaPy中使用方差衡量工作负载均衡？

核心思路 #

要实现负载均衡的方差优化，需分两步：定义单个用户负载计算逻辑，通过OptaPy约束流全局计算平均负载并最小化方差，求解完成后也可手动验证结果。

1. 实体类定义（补充负载计算方法） #

先确保User类能正确计算自身负载，Task类关联分配的用户：

from optapy import planning_entity, planning_id, planning_variable, value_range_provider

@planning_entity
class User:
    def __init__(self, id, fixed_cost, capacity):
        self.id = id
        self.fixed_cost = fixed_cost
        self.capacity = capacity
        self.tasks = []  # 存储分配给该用户的任务

    @planning_id
    def get_id(self):
        return self.id

    # 计算单个用户的工作负载
    def get_workload(self):
        total_task_cost = sum(task.cost for task in self.tasks)
        return (self.fixed_cost + total_task_cost) / self.capacity

@planning_entity
class Task:
    def __init__(self, id, cost):
        self.id = id
        self.cost = cost
        self.user = None  # 待分配的用户

    @planning_id
    def get_id(self):
        return self.id

    @planning_variable(User, value_range_provider_refs=["userRange"])
    def get_user(self):
        return self.user

    def set_user(self, user):
        self.user = user

2. 约束流定义：最小化负载方差 #

用OptaPy的约束流实现全局方差计算，将方差作为软约束惩罚，让求解器自动优化负载均衡：

from optapy import constraint_provider, HardSoftScore
from optapy.constraint import ConstraintFactory

@constraint_provider
def define_constraints(constraint_factory: ConstraintFactory):
    return [
        # 硬约束：确保所有任务都分配了用户（根据你的业务需求调整）
        constraint_factory.for_each(Task)
                          .filter(lambda task: task.user is None)
                          .penalize("未分配任务", HardSoftScore.ONE_HARD)
                          .as_constraint(),
        # 软约束：最小化负载方差
        minimize_workload_variance(constraint_factory)
    ]

def minimize_workload_variance(constraint_factory: ConstraintFactory):
    # 全局计算所有用户负载总和
    total_workload = constraint_factory.for_each(User)
                                       .map(lambda user: user.get_workload())
                                       .sum()
    # 全局计算用户总数
    user_count = constraint_factory.for_each(User).count()
    # 计算每个用户负载与平均值的差的平方和，作为惩罚项（值越小方差越小）
    return constraint_factory.for_each(User)
                            .join(total_workload)
                            .join(user_count)
                            .map(lambda user, total, count: (user.get_workload() - (total / count)) ** 2)
                            .sum()
                            .penalize("负载方差", HardSoftScore.ONE_SOFT)
                            .as_constraint()

3. 求解后验证平均负载与方差 #

求解得到最优解后，可手动遍历用户计算实际的平均负载和方差，用于结果验证：

# 假设solution是求解后的方案实例，需根据你的Solution类获取用户列表
users = solution.get_users()
workloads = [user.get_workload() for user in users]
average_workload = sum(workloads) / len(workloads)
variance = sum((w - average_workload) ** 2 for w in workloads) / len(workloads)

print(f"平均负载: {average_workload:.2f}")
print(f"负载方差: {variance:.4f}")

注意事项 #

• 如果需要保留原有的"最低负载优先"逻辑，可将其设为硬约束，再叠加方差最小化的软约束，实现多目标优化。
• get_workload方法需严格匹配你业务中的负载计算公式，确保数据准确。

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