简介一份面向边缘计算学习者的完整项目资源基于Python实现多用户单窃听者移动边缘计算MEC网络的卸载优化与资源优化结合深度强化学习DQN与凸优化方法适合用作毕设、课程设计或工程实训。资源包共51个文件包含16个Python脚本环境构建、DQN与DDPG算法实现、3个Jupyter Notebook演示、18个Excel实验数据文件及若干结果图表整体压缩包约211KB目录清晰便于复现。资源展示了在窃听者共谋与非共谋场景下的对比实验DQN在100轮训练后已明显优于全本地与全卸载策略并趋于收敛读者可据此理解强化学习在网络资源调度中的落地方式。目前已有162人学习下载适合希望掌握边缘计算、网络优化及深度强化学习结合应用的进阶学习者。1. 边缘不边基于 Python 的卸载优化与资源优化该怎么做你负责的园区边缘网关接入 30 路 AI 摄像头每帧画面 4MB云端推理往返 200ms本地 GPU 又被其他业务占掉一半。把所有任务全传云端带宽先堵全留在本地计算又排队。真正要解决的不是单一的网络问题而是“卸载到哪、分配多少资源”的组合优化。基于 Python 做边缘计算网络的卸载优化及资源优化就是用代码把这种组合决策变成可计算的模型再通过算法自动决定任务走向和节点算力配置。这个方向适合想进入边缘计算调度、边缘 AI 平台研发以及需要落地任务卸载策略的工程师。下文从建模开始逐步给出可运行的 Python 基线、MILP 最优解和基于 LSTM 的动态资源调整方法。2. 把卸载优化建模成 Python 可计算的优化对象2.1 先搞清楚卸载优化不是简单的传给边缘服务器很多刚接触边缘计算的开发者会把卸载优化理解成“网络好就传云端网络差就本地算”但真实边缘网络里的卸载决策至少包含三层任务是否卸载、卸载到哪个节点、每个节点分配多少 CPU 或带宽资源。只做前两层边缘节点会被突发任务打满任务排队时延反而超过本地执行。我一般会先把问题抽象成“时延 能耗”的加权目标。任务 i 在本地执行时主要成本是计算量 / 本地CPU频率卸载到边缘时成本变成数据量 / 带宽 计算量 / 边缘CPU频率同时还要计入传输能耗与计算能耗。这个模型足够表达边缘计算网络的核心矛盾卸载能降低本地压力但会引入传输开销资源分配能缩短计算耗时但要避免节点过载。2.2 用 dataclass 定义任务与节点让模型可直接跑起来在 Python 里做边缘计算网络的卸载优化第一步不是写算法而是把任务、节点和成本函数定义清楚。我习惯用dataclass描述一个任务避免后面传太多参数。你可以把这个定义理解为模型的“最小公共接口”。from dataclasses import dataclass dataclass class EdgeTask: arrival_time: float # 任务到达时刻秒 data_size: float # 数据传输量MB cycles_per_mb: float # 每 MB 需要多少 CPU 周期cycles/MB deadline: float # 相对截止时间秒 local_rate: float # 本地可用计算速率cycles/s edge_rate: float # 边缘服务器可用计算速率cycles/s def compute_cost(task: EdgeTask, use_edge: bool): # 本地执行 if not use_edge: t_calc task.data_size * task.cycles_per_mb / task.local_rate e_calc task.data_size * task.cycles_per_mb * 2.4e-8 return t_calc, e_calc # 卸载到边缘传输 边缘计算 bw_mbps 10.0 t_trans task.data_size / bw_mbps t_calc task.data_size * task.cycles_per_mb / task.edge_rate e_trans task.data_size * 18.5 # 传输能耗系数单位可调 e_calc task.data_size * task.cycles_per_mb * 1.2e-8 return t_trans t_calc, e_trans e_calc这段代码里local_rate和edge_rate的单位是 cycles/s比如本地 CPU 为 2GHz就填2e9。cycles_per_mb表示处理 1MB 数据需要的 CPU 周期数这个值通常由离线基准测试得到。compute_cost返回(时延秒, 能耗单位)后续所有算法都基于这个接口计算。注意能耗系数不是硬件厂商给定的而是我对实际边缘网关功耗做简化后的估计值本地 CPU 功耗密度高于边缘设备卸载到边缘虽然网络模块耗电但计算功耗更低。你可以把这两个系数写到配置表里真机验证时替换成实测值。2.3 优化目标与约束用符号表把问题讲清楚模型里需要用到一批符号我通常会在代码文件头部注释中列出方便团队评审时对齐口径。下表是常见参数符号含义示例值D_i任务数据大小2 MBC_i每 MB 所需 CPU 周期100 kcycles/MBf_loc / f_edge本地/边缘计算速率2e9 / 8e9 cycles/sB上行带宽10 MbpsT_loc / T_edge本地/卸载时延计算得出E_loc / E_edge本地/卸载能耗计算得出alpha时延权重0~10.5对于一批任务卸载优化的目标可以写成最小化alpha * 总时延 (1 - alpha) * 总能耗。alpha 越大越偏重低时延alpha 越小越偏重低能耗。约束条件是边缘节点在任意时间窗口内的累计计算量不超过其处理能力或者更简单地在后续 MILP 模型中用“额外能耗预算”约束来控制卸载到边缘的任务量。这里的能量参数和时间参数处于不同量纲直接加权会导致结果失真。常见做法是在比较T和E之前做 min-max 归一化后续章节的代码里我会演示一个轻量版。3. 用 Python 实现卸载决策贪心与 MILP3.1 生成模拟任务流泊松到达与随机负载做边缘计算网络的卸载优化第一步是获得一批可复现的任务。我经常用泊松过程模拟业务到达平均每 50ms 来一个任务持续生成 1000 个任务然后为每个任务随机指定数据大小和计算密度。import numpy as np rng np.random.default_rng(42) arrivals np.cumsum(rng.exponential(scale0.05, size1000)) tasks [] for i in range(1000): tasks.append(EdgeTask( arrival_timearrivals[i], data_sizefloat(rng.uniform(1, 5)), # 1~5 MB cycles_per_mbfloat(rng.uniform(50e3, 200e3)), # 50k~200k cycles/MB deadlinefloat(rng.uniform(0.08, 0.2)), # 80~200ms 必须完成 local_rate2e9, edge_rate8e9, ))参数说明rng.exponential生成指数间隔scale0.05表示平均间隔 50ms。local_rate2e9对应 2GHz CPUedge_rate8e9对应一台 8 核边缘服务器的可用算力。计算密度覆盖了从简单数据清洗到图像预处理的典型范围。需要更真实的数据时可以换成你集群上的任务历史到达间隔用np.random.choice采样即可。这里生成的 1000 个任务会被后续贪心和 MILP 算法使用。注意你需要在专家加裁一个场景比如总任务量超过边缘处理能力才能看出卸载优化算法的差异。3.2 贪心卸载可落地的快速决策函数贪心策略的特点是每个任务独立决策不考虑全局影响。代码里比较本地与边缘的加权得分得分较低的执行。def greedy_offload(tasks, alpha0.5): decisions [] total_delay 0.0 total_energy 0.0 for task in tasks: t_loc, e_loc compute_cost(task, use_edgeFalse) t_edge, e_edge compute_cost(task, use_edgeTrue) # 简单归一化能耗统一除以一个量级常数 score_loc alpha * t_loc (1 - alpha) * e_loc / 10 score_edge alpha * t_edge (1 - alpha) * e_edge / 10 use_edge score_edge score_loc decisions.append(use_edge) total_delay t_edge if use_edge else t_loc total_energy e_edge if use_edge else e_loc return decisions, total_delay, total_energy代码逻辑很简单对每个任务分别调用compute_cost再把时延和能耗加权。e_loc/10是为了让能耗数值落在和时延相近的量级避免时延权重被能耗淹没。alpha是策略参数工程上常见做法是先离线跑几组不同 alpha 的任务流画出时延和能耗的帕累托曲线再选择一个业务可接受的点。贪心的缺点是它会忽略边缘节点总容量。当 1000 个任务都判断“边缘更快”时决策全部为卸载边缘 CPU 排队会让真实时延远高于模型预测。因此贪心更适合作为上界或者用于单节点、任务量远小于节点容量的场景。3.3 MILP在滚动窗口里拿到全局最优基线为了评估贪心的损失我通常会在一个小窗口比如 50 个任务内用混合整数线性规划求解最优卸载集。Python 里直接用scipy.optimize.milp不需要引入额外的优化器依赖。思路是全部任务本地执行作为基准用 0-1 变量表示是否卸载到边缘最小化加权的时延增量和能耗增量同时限制新增的能耗不能超过预算。from scipy.optimize import milp, LinearConstraint, Bounds import numpy as np def solve_milp(tasks, alpha0.5, energy_budget_ratio1.2): n len(tasks) base_delay 0.0 base_energy 0.0 t_delta [] e_delta [] for task in tasks: t_loc, e_loc compute_cost(task, use_edgeFalse) t_edge, e_edge compute_cost(task, use_edgeTrue) base_delay t_loc base_energy e_loc t_delta.append(t_edge - t_loc) e_delta.append(e_edge - e_loc) # 目标min sum(alpha*t_delta (1-alpha)*e_delta) c alpha * np.array(t_delta) (1 - alpha) * np.array(e_delta) # 约束卸载带来的额外能耗 预算 energy_budget base_energy * (energy_budget_ratio - 1.0) A np.array(e_delta).reshape(1, -1) constraints [LinearConstraint(A, -np.inf, energy_budget)] result milp(c, integralitynp.ones(n), boundsBounds(0, 1), constraintsconstraints) decisions result.x 0.5 total_delay base_delay sum( t_delta[i] for i, d in enumerate(decisions) if d ) total_energy base_energy sum( e_delta[i] for i, d in enumerate(decisions) if d ) return decisions, total_delay, total_energy参数说明energy_budget_ratio1.2表示允许比全本地执行多消耗 20% 的能量用来换取时延降低。integralitynp.ones(n)强制 0-1 整数。milp返回的是 SCIPY OptimizeResultresult.x是每个任务的卸载标志。运行一次小型对比你会看到贪心在耗时上略高但能耗明显更低MILP 则是给定能耗预算下的时延最优。实际边缘网络里任务是持续到达的不可能对全量任务做 MILP。我一般每 5 秒滚动一次每次只规划未来 50 个任务然后把决策前传给调度器。MILP 的解作为离线评估指标用于判断贪心策略距离最优还有多少差距。4. 时变环境下用 LSTM 预测负载并动态分配资源4.1 静态决策为什么在真实边缘网络里不够用前面的模型假设边缘节点资源固定任务到达率稳定但真实边缘网络常常出现突发流白天业务高峰任务到达率翻倍夜间大幅下降。如果edge_rate一直固定在 8e9高峰时边缘节点积压低峰时算力闲置。资源优化的前端应该是“预测下一段时间的负载”后端再根据预测值动态调整分配给每个节点的计算资源。这不是锦上添花而是边缘计算网络资源优化的常规做法。预测结果用于两个地方一是调整边缘节点的虚拟 CPU 配额二是改变卸载决策中的带宽分配。Python 生态里做时间序列预测最顺手的是 PyTorch LSTM哪怕只用一个隐藏层也能捕捉到达率的长短周期。4.2 用 PyTorch 构建到达率预测模型先构造一个简单的到达率预测网络输入是过去 24 个时间窗口的任务数输出是下一个窗口的预测值。import torch import torch.nn as nn class ArrivalPredictor(nn.Module): def __init__(self, input_size1, hidden_size16, num_layers1): super().__init__() self.lstm nn.LSTM( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue, ) self.fc nn.Linear(hidden_size, 1) def forward(self, x): # x: (batch, window, input_size) out, _ self.lstm(x) # out: (batch, window, hidden) return self.fc(out[:, -1, :]) # 取最后一个时间步的输出训练数据可以用历史到达记录构造。假设我们把时间切成等长窗口比如每 5 秒统计一次任务数那么window24表示用最近 2 分钟的到达量预测未来 5 秒。下面是构造训练样本的核心逻辑def make_samples(counts, window24): X, y [], [] for i in range(len(counts) - window): X.append(counts[i:i window]) y.append(counts[i window]) return torch.tensor(X, dtypetorch.float32).unsqueeze(-1), \ torch.tensor(y, dtypetorch.float32) # 假设 counts 是从日志里统计出的每 5 秒任务数 # X: (样本数, 24, 1)训练时用 MSELoss 和 Adam 优化器即可。我这里略去训练循环因为它和普通时间序列预测没有区别。注意对counts做归一化后再训练否则 LSTM 收敛很慢。实际代码里可以用sklearn.preprocessing.MinMaxScaler这是边缘计算资源优化里常见的数据预处理步骤不会破坏可复现性。4.3 让预测结果影响资源分配一个可解释的调整函数得到下一个时间窗口的到达率pred后需要把它转成边缘节点的资源配额。常见做法是维护一个基线频率base_edge_rate然后按照预测值与历史均值的偏离程度做线性调整def adjust_edge_rate(pred, hist_mean, base_edge_rate, k0.15): # pred: 预测的下窗口任务数 # hist_mean: 历史平均任务数 delta (pred - hist_mean) / (hist_mean 1e-6) new_rate base_edge_rate * (1.0 k * delta) return float(np.clip(new_rate, 0.5 * base_edge_rate, 1.5 * base_edge_rate))参数说明k是调整强度k0.15表示预测值比平均值高 10% 时边缘算力提高 1.5%。np.clip把资源变化限制在正负 50%防止某个瞬间预测尖峰导致资源分配抖得厉害。把调整后的edge_rate传给第三章的compute_cost和greedy_offload就能看到新的总时延和能耗。这种做法的好处是逻辑简单、可解释维护人员可以直接看出“预测任务数增长了 20%所以边缘 CPU 配额提高了 3%”。如果你希望更精细化可以用同样的思路动态调整带宽bw_mbps但注意带宽调整主要影响传输时延对本地任务没有作用。5. 卸载优化落地前值得常备的 3 个 Python 技巧5.1 用 numpy 向量化替代逐任务 for 循环第 3 章里的贪心函数用 for 循环处理任务代码直观但性能一般。如果任务流每秒上千条建议把compute_cost改成向量化版本。所有任务的数据大小和周期可以放入一维数组本地和边缘的时延同时计算data np.array([t.data_size for t in tasks]) cycles np.array([t.cycles_per_mb for t in tasks]) local_rate tasks[0].local_rate edge_rate tasks[0].edge_rate t_loc data * cycles / local_rate t_edge data / bw_mbps data * cycles / edge_rate # 用布尔掩码统一决策这种方式几乎不改变原逻辑却能减少 80% 以上的 Python 循环开销。5.2 大窗口 MILP 太慢时用 numba 加速贪心滚动窗口若取到 200 个任务MILP 可能耗时数百毫秒在 5 秒级的调度周期里还能接受。但如果窗口更大或节点更多建议用numba把贪心决策编译成机器码。你只需要给函数加上njit装饰器并把内部用到的小列表改成 numpy 数组。需要提醒的是compute_cost里的 Python 对象和循环必须全部抽离成基本数值数组numba才能正常工作。5.3 调权重之前先做 min-max 归一化alpha 的取值看起来是 0 到 1但时延通常以毫秒计能耗以焦耳计两者直接乘权重会偏向数值大的项。更稳的做法是先把所有任务的本地/边缘时延和能耗做 min-max 归一化得出 0~1 的分数后再用 alpha 加权。这样 alpha 的含义就稳定成为“时延相对重要程度”更换硬件后不需要重新调 alpha。实际运行时归一化的均值可以离线统计在线只做减均值和除标准差两步开销很小。本文还有配套的精品资源点击获取 Python实现边缘计算卸载优化与资源动态分配实战
简介一份面向边缘计算学习者的完整项目资源基于Python实现多用户单窃听者移动边缘计算MEC网络的卸载优化与资源优化结合深度强化学习DQN与凸优化方法适合用作毕设、课程设计或工程实训。资源包共51个文件包含16个Python脚本环境构建、DQN与DDPG算法实现、3个Jupyter Notebook演示、18个Excel实验数据文件及若干结果图表整体压缩包约211KB目录清晰便于复现。资源展示了在窃听者共谋与非共谋场景下的对比实验DQN在100轮训练后已明显优于全本地与全卸载策略并趋于收敛读者可据此理解强化学习在网络资源调度中的落地方式。目前已有162人学习下载适合希望掌握边缘计算、网络优化及深度强化学习结合应用的进阶学习者。1. 边缘不边基于 Python 的卸载优化与资源优化该怎么做你负责的园区边缘网关接入 30 路 AI 摄像头每帧画面 4MB云端推理往返 200ms本地 GPU 又被其他业务占掉一半。把所有任务全传云端带宽先堵全留在本地计算又排队。真正要解决的不是单一的网络问题而是“卸载到哪、分配多少资源”的组合优化。基于 Python 做边缘计算网络的卸载优化及资源优化就是用代码把这种组合决策变成可计算的模型再通过算法自动决定任务走向和节点算力配置。这个方向适合想进入边缘计算调度、边缘 AI 平台研发以及需要落地任务卸载策略的工程师。下文从建模开始逐步给出可运行的 Python 基线、MILP 最优解和基于 LSTM 的动态资源调整方法。2. 把卸载优化建模成 Python 可计算的优化对象2.1 先搞清楚卸载优化不是简单的传给边缘服务器很多刚接触边缘计算的开发者会把卸载优化理解成“网络好就传云端网络差就本地算”但真实边缘网络里的卸载决策至少包含三层任务是否卸载、卸载到哪个节点、每个节点分配多少 CPU 或带宽资源。只做前两层边缘节点会被突发任务打满任务排队时延反而超过本地执行。我一般会先把问题抽象成“时延 能耗”的加权目标。任务 i 在本地执行时主要成本是计算量 / 本地CPU频率卸载到边缘时成本变成数据量 / 带宽 计算量 / 边缘CPU频率同时还要计入传输能耗与计算能耗。这个模型足够表达边缘计算网络的核心矛盾卸载能降低本地压力但会引入传输开销资源分配能缩短计算耗时但要避免节点过载。2.2 用 dataclass 定义任务与节点让模型可直接跑起来在 Python 里做边缘计算网络的卸载优化第一步不是写算法而是把任务、节点和成本函数定义清楚。我习惯用dataclass描述一个任务避免后面传太多参数。你可以把这个定义理解为模型的“最小公共接口”。from dataclasses import dataclass dataclass class EdgeTask: arrival_time: float # 任务到达时刻秒 data_size: float # 数据传输量MB cycles_per_mb: float # 每 MB 需要多少 CPU 周期cycles/MB deadline: float # 相对截止时间秒 local_rate: float # 本地可用计算速率cycles/s edge_rate: float # 边缘服务器可用计算速率cycles/s def compute_cost(task: EdgeTask, use_edge: bool): # 本地执行 if not use_edge: t_calc task.data_size * task.cycles_per_mb / task.local_rate e_calc task.data_size * task.cycles_per_mb * 2.4e-8 return t_calc, e_calc # 卸载到边缘传输 边缘计算 bw_mbps 10.0 t_trans task.data_size / bw_mbps t_calc task.data_size * task.cycles_per_mb / task.edge_rate e_trans task.data_size * 18.5 # 传输能耗系数单位可调 e_calc task.data_size * task.cycles_per_mb * 1.2e-8 return t_trans t_calc, e_trans e_calc这段代码里local_rate和edge_rate的单位是 cycles/s比如本地 CPU 为 2GHz就填2e9。cycles_per_mb表示处理 1MB 数据需要的 CPU 周期数这个值通常由离线基准测试得到。compute_cost返回(时延秒, 能耗单位)后续所有算法都基于这个接口计算。注意能耗系数不是硬件厂商给定的而是我对实际边缘网关功耗做简化后的估计值本地 CPU 功耗密度高于边缘设备卸载到边缘虽然网络模块耗电但计算功耗更低。你可以把这两个系数写到配置表里真机验证时替换成实测值。2.3 优化目标与约束用符号表把问题讲清楚模型里需要用到一批符号我通常会在代码文件头部注释中列出方便团队评审时对齐口径。下表是常见参数符号含义示例值D_i任务数据大小2 MBC_i每 MB 所需 CPU 周期100 kcycles/MBf_loc / f_edge本地/边缘计算速率2e9 / 8e9 cycles/sB上行带宽10 MbpsT_loc / T_edge本地/卸载时延计算得出E_loc / E_edge本地/卸载能耗计算得出alpha时延权重0~10.5对于一批任务卸载优化的目标可以写成最小化alpha * 总时延 (1 - alpha) * 总能耗。alpha 越大越偏重低时延alpha 越小越偏重低能耗。约束条件是边缘节点在任意时间窗口内的累计计算量不超过其处理能力或者更简单地在后续 MILP 模型中用“额外能耗预算”约束来控制卸载到边缘的任务量。这里的能量参数和时间参数处于不同量纲直接加权会导致结果失真。常见做法是在比较T和E之前做 min-max 归一化后续章节的代码里我会演示一个轻量版。3. 用 Python 实现卸载决策贪心与 MILP3.1 生成模拟任务流泊松到达与随机负载做边缘计算网络的卸载优化第一步是获得一批可复现的任务。我经常用泊松过程模拟业务到达平均每 50ms 来一个任务持续生成 1000 个任务然后为每个任务随机指定数据大小和计算密度。import numpy as np rng np.random.default_rng(42) arrivals np.cumsum(rng.exponential(scale0.05, size1000)) tasks [] for i in range(1000): tasks.append(EdgeTask( arrival_timearrivals[i], data_sizefloat(rng.uniform(1, 5)), # 1~5 MB cycles_per_mbfloat(rng.uniform(50e3, 200e3)), # 50k~200k cycles/MB deadlinefloat(rng.uniform(0.08, 0.2)), # 80~200ms 必须完成 local_rate2e9, edge_rate8e9, ))参数说明rng.exponential生成指数间隔scale0.05表示平均间隔 50ms。local_rate2e9对应 2GHz CPUedge_rate8e9对应一台 8 核边缘服务器的可用算力。计算密度覆盖了从简单数据清洗到图像预处理的典型范围。需要更真实的数据时可以换成你集群上的任务历史到达间隔用np.random.choice采样即可。这里生成的 1000 个任务会被后续贪心和 MILP 算法使用。注意你需要在专家加裁一个场景比如总任务量超过边缘处理能力才能看出卸载优化算法的差异。3.2 贪心卸载可落地的快速决策函数贪心策略的特点是每个任务独立决策不考虑全局影响。代码里比较本地与边缘的加权得分得分较低的执行。def greedy_offload(tasks, alpha0.5): decisions [] total_delay 0.0 total_energy 0.0 for task in tasks: t_loc, e_loc compute_cost(task, use_edgeFalse) t_edge, e_edge compute_cost(task, use_edgeTrue) # 简单归一化能耗统一除以一个量级常数 score_loc alpha * t_loc (1 - alpha) * e_loc / 10 score_edge alpha * t_edge (1 - alpha) * e_edge / 10 use_edge score_edge score_loc decisions.append(use_edge) total_delay t_edge if use_edge else t_loc total_energy e_edge if use_edge else e_loc return decisions, total_delay, total_energy代码逻辑很简单对每个任务分别调用compute_cost再把时延和能耗加权。e_loc/10是为了让能耗数值落在和时延相近的量级避免时延权重被能耗淹没。alpha是策略参数工程上常见做法是先离线跑几组不同 alpha 的任务流画出时延和能耗的帕累托曲线再选择一个业务可接受的点。贪心的缺点是它会忽略边缘节点总容量。当 1000 个任务都判断“边缘更快”时决策全部为卸载边缘 CPU 排队会让真实时延远高于模型预测。因此贪心更适合作为上界或者用于单节点、任务量远小于节点容量的场景。3.3 MILP在滚动窗口里拿到全局最优基线为了评估贪心的损失我通常会在一个小窗口比如 50 个任务内用混合整数线性规划求解最优卸载集。Python 里直接用scipy.optimize.milp不需要引入额外的优化器依赖。思路是全部任务本地执行作为基准用 0-1 变量表示是否卸载到边缘最小化加权的时延增量和能耗增量同时限制新增的能耗不能超过预算。from scipy.optimize import milp, LinearConstraint, Bounds import numpy as np def solve_milp(tasks, alpha0.5, energy_budget_ratio1.2): n len(tasks) base_delay 0.0 base_energy 0.0 t_delta [] e_delta [] for task in tasks: t_loc, e_loc compute_cost(task, use_edgeFalse) t_edge, e_edge compute_cost(task, use_edgeTrue) base_delay t_loc base_energy e_loc t_delta.append(t_edge - t_loc) e_delta.append(e_edge - e_loc) # 目标min sum(alpha*t_delta (1-alpha)*e_delta) c alpha * np.array(t_delta) (1 - alpha) * np.array(e_delta) # 约束卸载带来的额外能耗 预算 energy_budget base_energy * (energy_budget_ratio - 1.0) A np.array(e_delta).reshape(1, -1) constraints [LinearConstraint(A, -np.inf, energy_budget)] result milp(c, integralitynp.ones(n), boundsBounds(0, 1), constraintsconstraints) decisions result.x 0.5 total_delay base_delay sum( t_delta[i] for i, d in enumerate(decisions) if d ) total_energy base_energy sum( e_delta[i] for i, d in enumerate(decisions) if d ) return decisions, total_delay, total_energy参数说明energy_budget_ratio1.2表示允许比全本地执行多消耗 20% 的能量用来换取时延降低。integralitynp.ones(n)强制 0-1 整数。milp返回的是 SCIPY OptimizeResultresult.x是每个任务的卸载标志。运行一次小型对比你会看到贪心在耗时上略高但能耗明显更低MILP 则是给定能耗预算下的时延最优。实际边缘网络里任务是持续到达的不可能对全量任务做 MILP。我一般每 5 秒滚动一次每次只规划未来 50 个任务然后把决策前传给调度器。MILP 的解作为离线评估指标用于判断贪心策略距离最优还有多少差距。4. 时变环境下用 LSTM 预测负载并动态分配资源4.1 静态决策为什么在真实边缘网络里不够用前面的模型假设边缘节点资源固定任务到达率稳定但真实边缘网络常常出现突发流白天业务高峰任务到达率翻倍夜间大幅下降。如果edge_rate一直固定在 8e9高峰时边缘节点积压低峰时算力闲置。资源优化的前端应该是“预测下一段时间的负载”后端再根据预测值动态调整分配给每个节点的计算资源。这不是锦上添花而是边缘计算网络资源优化的常规做法。预测结果用于两个地方一是调整边缘节点的虚拟 CPU 配额二是改变卸载决策中的带宽分配。Python 生态里做时间序列预测最顺手的是 PyTorch LSTM哪怕只用一个隐藏层也能捕捉到达率的长短周期。4.2 用 PyTorch 构建到达率预测模型先构造一个简单的到达率预测网络输入是过去 24 个时间窗口的任务数输出是下一个窗口的预测值。import torch import torch.nn as nn class ArrivalPredictor(nn.Module): def __init__(self, input_size1, hidden_size16, num_layers1): super().__init__() self.lstm nn.LSTM( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue, ) self.fc nn.Linear(hidden_size, 1) def forward(self, x): # x: (batch, window, input_size) out, _ self.lstm(x) # out: (batch, window, hidden) return self.fc(out[:, -1, :]) # 取最后一个时间步的输出训练数据可以用历史到达记录构造。假设我们把时间切成等长窗口比如每 5 秒统计一次任务数那么window24表示用最近 2 分钟的到达量预测未来 5 秒。下面是构造训练样本的核心逻辑def make_samples(counts, window24): X, y [], [] for i in range(len(counts) - window): X.append(counts[i:i window]) y.append(counts[i window]) return torch.tensor(X, dtypetorch.float32).unsqueeze(-1), \ torch.tensor(y, dtypetorch.float32) # 假设 counts 是从日志里统计出的每 5 秒任务数 # X: (样本数, 24, 1)训练时用 MSELoss 和 Adam 优化器即可。我这里略去训练循环因为它和普通时间序列预测没有区别。注意对counts做归一化后再训练否则 LSTM 收敛很慢。实际代码里可以用sklearn.preprocessing.MinMaxScaler这是边缘计算资源优化里常见的数据预处理步骤不会破坏可复现性。4.3 让预测结果影响资源分配一个可解释的调整函数得到下一个时间窗口的到达率pred后需要把它转成边缘节点的资源配额。常见做法是维护一个基线频率base_edge_rate然后按照预测值与历史均值的偏离程度做线性调整def adjust_edge_rate(pred, hist_mean, base_edge_rate, k0.15): # pred: 预测的下窗口任务数 # hist_mean: 历史平均任务数 delta (pred - hist_mean) / (hist_mean 1e-6) new_rate base_edge_rate * (1.0 k * delta) return float(np.clip(new_rate, 0.5 * base_edge_rate, 1.5 * base_edge_rate))参数说明k是调整强度k0.15表示预测值比平均值高 10% 时边缘算力提高 1.5%。np.clip把资源变化限制在正负 50%防止某个瞬间预测尖峰导致资源分配抖得厉害。把调整后的edge_rate传给第三章的compute_cost和greedy_offload就能看到新的总时延和能耗。这种做法的好处是逻辑简单、可解释维护人员可以直接看出“预测任务数增长了 20%所以边缘 CPU 配额提高了 3%”。如果你希望更精细化可以用同样的思路动态调整带宽bw_mbps但注意带宽调整主要影响传输时延对本地任务没有作用。5. 卸载优化落地前值得常备的 3 个 Python 技巧5.1 用 numpy 向量化替代逐任务 for 循环第 3 章里的贪心函数用 for 循环处理任务代码直观但性能一般。如果任务流每秒上千条建议把compute_cost改成向量化版本。所有任务的数据大小和周期可以放入一维数组本地和边缘的时延同时计算data np.array([t.data_size for t in tasks]) cycles np.array([t.cycles_per_mb for t in tasks]) local_rate tasks[0].local_rate edge_rate tasks[0].edge_rate t_loc data * cycles / local_rate t_edge data / bw_mbps data * cycles / edge_rate # 用布尔掩码统一决策这种方式几乎不改变原逻辑却能减少 80% 以上的 Python 循环开销。5.2 大窗口 MILP 太慢时用 numba 加速贪心滚动窗口若取到 200 个任务MILP 可能耗时数百毫秒在 5 秒级的调度周期里还能接受。但如果窗口更大或节点更多建议用numba把贪心决策编译成机器码。你只需要给函数加上njit装饰器并把内部用到的小列表改成 numpy 数组。需要提醒的是compute_cost里的 Python 对象和循环必须全部抽离成基本数值数组numba才能正常工作。5.3 调权重之前先做 min-max 归一化alpha 的取值看起来是 0 到 1但时延通常以毫秒计能耗以焦耳计两者直接乘权重会偏向数值大的项。更稳的做法是先把所有任务的本地/边缘时延和能耗做 min-max 归一化得出 0~1 的分数后再用 alpha 加权。这样 alpha 的含义就稳定成为“时延相对重要程度”更换硬件后不需要重新调 alpha。实际运行时归一化的均值可以离线统计在线只做减均值和除标准差两步开销很小。本文还有配套的精品资源点击获取