第67课 - 任务调度问题
学习目标
- 理解任务调度问题:掌握任务调度问题的定义、目标和应用场景
- 掌握贪心策略:学会如何运用贪心算法解决带截止时间的任务调度问题
- 实现贪心算法:能够编写代码实现任务调度算法
- 分析算法复杂度:理解算法的时间复杂度和空间复杂度
- 应用扩展:了解任务调度问题的变种和优化方法
核心概念
1. 问题定义
任务调度问题是一个经典的贪心算法应用场景。给定一组任务,每个任务有:
- 执行时间:完成任务需要的时间(本课假设每个任务执行时间为1个时间单位)
- 截止时间:任务必须在此时间点之前完成
- 利润:完成任务能获得的收益
我们的目标是在给定的时间范围内,选择一组任务来执行,使得总利润最大。
2. 贪心策略
核心思想:优先选择利润最高的任务,尽可能在截止时间前完成它。
为什么这个策略有效?因为:
- 高利润任务对总收益贡献大
- 通过优先安排高利润任务,我们确保了收益的最大化
- 如果高利润任务无法按时完成,再考虑次高利润的任务
3. 时间槽管理
我们可以用一个数组来表示时间槽(time slots),每个时间槽只能安排一个任务。从任务的截止时间开始向前寻找空闲的时间槽。
例子: 假设我们有3个时间槽 [1, 2, 3],任务A截止时间为2,任务B截止时间为3。
- 先安排利润高的任务A:时间槽2被占用
- 再安排任务B:检查时间槽3,空闲则占用
代码示例
class Job:
def __init__(self, id, deadline, profit):
self.id = id
self.deadline = deadline # 截止时间
self.profit = profit # 利润
def __repr__(self):
return f"任务{self.id}(截止:{self.deadline}, 利润:{self.profit})"
def schedule_jobs(jobs):
"""
任务调度算法 - 最大化总利润
参数:
jobs: 任务列表,每个任务是一个Job对象
返回:
(scheduled_jobs, total_profit):
scheduled_jobs - 被安排执行的任务列表
total_profit - 总利润
"""
# 1. 按利润降序排序任务
jobs_sorted = sorted(jobs, key=lambda x: x.profit, reverse=True)
# 2. 找出最大截止时间,确定时间槽数量
max_deadline = max(job.deadline for job in jobs)
# 3. 初始化时间槽数组,False表示空闲,True表示已占用
time_slots = [False] * (max_deadline + 1) # 索引0不使用,从1开始
# 4. 安排任务
scheduled_jobs = []
total_profit = 0
for job in jobs_sorted:
# 从截止时间开始向前寻找空闲时间槽
for slot in range(min(max_deadline, job.deadline), 0, -1):
if not time_slots[slot]: # 找到空闲时间槽
# 安排任务
time_slots[slot] = True
scheduled_jobs.append(job)
total_profit += job.profit
break
return scheduled_jobs, total_profit
# 示例运行
if __name__ == "__main__":
# 创建任务列表
jobs = [
Job(1, 2, 100), # 任务1:截止时间2,利润100
Job(2, 1, 19), # 任务2:截止时间1,利润19
Job(3, 2, 27), # 任务3:截止时间2,利润27
Job(4, 1, 25), # 任务4:截止时间1,利润25
Job(5, 3, 15), # 任务5:截止时间3,利润15
]
print("原始任务列表:")
for job in jobs:
print(f" {job}")
# 执行调度算法
scheduled, total = schedule_jobs(jobs)
print("\n调度结果:")
print(f"被安排的任务: {scheduled}")
print(f"总利润: {total}")
# 显示时间槽安排
print("\n时间槽安排:")
max_deadline = max(job.deadline for job in jobs)
time_slots = ["空"] * (max_deadline + 1)
# 重新模拟调度过程来显示安排
jobs_sorted = sorted(jobs, key=lambda x: x.profit, reverse=True)
temp_slots = [False] * (max_deadline + 1)
for job in jobs_sorted:
for slot in range(min(max_deadline, job.deadline), 0, -1):
if not temp_slots[slot]:
temp_slots[slot] = True
time_slots[slot] = f"任务{job.id}"
break
for i in range(1, max_deadline + 1):
print(f" 时间槽{i}: {time_slots[i]}")
代码输出:
原始任务列表:
任务1(截止:2, 利润:100)
任务2(截止:1, 利润:19)
任务3(截止:2, 利润:27)
任务4(截止:1, 利润:25)
任务5(截止:3, 利润:15)
调度结果:
被安排的任务: [任务1(截止:2, 利润:100), 任务4(截止:1, 利润:25), 任务5(截止:3, 利润:15)]
总利润: 140
时间槽安排:
时间槽1: 任务4
时间槽2: 任务1
时间槽3: 任务5
实践练习
练习1:基础任务调度
要求:给定以下任务,找出最大利润的任务调度方案。
任务列表:
- 任务A:截止时间1,利润50
- 任务B:截止时间2,利润30
- 任务C:截止时间3,利润40
- 任务D:截止时间2,利润20
预期输出:
调度的任务: [A, C, B]
总利润: 120
练习2:扩展任务调度
要求:修改上面的代码,当有多个相同利润的任务时,优先安排截止时间更早的任务。
提示:修改排序函数,让排序条件变为(-profit, deadline)。
测试数据:
- 任务X:截止时间3,利润100
- 任务Y:截止时间2,利润100
- 任务Z:截止时间1,利润50
练习3:并查集优化(进阶)
要求:研究并实现使用并查集(Disjoint Set Union)优化任务调度算法,将时间复杂度从O(n²)降低到O(n log n)。
提示:并查集可以快速找到每个任务应该安排的时间槽,避免线性扫描。
常见错误
1. 排序策略错误
错误:按截止时间排序而不是按利润排序
# 错误做法
jobs_sorted = sorted(jobs, key=lambda x: x.deadline) # 这样会导致错过高利润任务
2. 时间槽管理不当
错误:没有正确处理时间槽的索引问题
# 错误:索引从0开始,但截止时间从1开始
time_slots = [False] * max_deadline # 应该是 max_deadline + 1
3. 边界条件处理错误
错误:没有考虑截止时间可能超过总时间槽的情况
# 错误:直接使用job.deadline作为索引
for slot in range(job.deadline, 0, -1): # 可能越界
# 正确:限制在最大时间槽范围内
for slot in range(min(max_deadline, job.deadline), 0, -1):
4. 贪心选择不彻底
错误:只考虑当前任务,不考虑全局最优
# 错误:只安排第一个能安排的任务
for job in jobs:
if time_slots[job.deadline] is False:
# 安排任务
正确做法:先按利润排序,确保总是优先安排高利润任务。
小结
关键要点回顾
-
问题本质:任务调度问题是在有限的时间槽内,选择最大化总利润的任务子集。
-
贪心策略:按利润降序排序任务是关键,这确保了我们总是优先尝试安排最高利润的任务。
-
时间槽管理:使用数组标记时间槽的占用情况,从截止时间向前寻找空闲槽位。
-
算法步骤:
- 按利润降序排序任务
- 找出最大截止时间,确定时间槽数量
- 遍历排序后的任务,尝试安排每个任务
- 返回被安排的任务和总利润
-
时间复杂度:
- 排序:O(n log n)
- 安排任务:最坏情况下O(n²),平均O(n log n)
- 总体复杂度:O(n²)(基础实现),可通过并查集优化到O(n log n)
-
应用场景:
- 操作系统进程调度
- 云计算任务调度
- 项目管理任务分配
- 生产排程问题
拓展思考
- 如果任务的执行时间不为1,问题会如何变化?
- 如果允许任务延迟完成但要扣除利润,该如何建模?
- 如何处理任务之间存在依赖关系的情况?
掌握任务调度问题不仅能帮助我们理解贪心算法的应用,还能为解决实际工程中的调度问题提供思路。在下一课中,我们将深入探讨贪心算法的正确性证明方法。