Python作业调度算法实现：资源冲突处理与高效调度咨询

核心设计逻辑 #

要满足你的调度需求，核心要做好这几点：

• 用资源-时间映射表快速追踪每个资源的占用状态，避免低效遍历所有已调度作业
• 对冲突作业采用轮询重试队列，确保没有作业被遗漏
• 开启防重叠时，优先保证资源不冲突，允许作业启动时间超出原设定的interval

具体实现代码 #

class JobScheduler:
    def __init__(self):
        # 存储每个资源的最近结束时间，key: 资源ID, value: 时间戳/时长刻度
        self.resource_last_end = {}
        # 最终调度结果，格式: (作业ID, 资源ID, 启动时间, 结束时间)
        self.scheduled = []

    def schedule_jobs(self, jobs):
        # 待重试队列，存放暂时无法调度的冲突作业
        retry_queue = jobs.copy()
        
        while retry_queue:
            job = retry_queue.pop(0)
            job_id = job['id']
            resource = job['resource']
            duration = job['duration']
            interval = job.get('interval', 0)
            prevent_overlap = job.get('prevent_resource_overlap', True)
            
            # 作业的初始计划启动时间（基于interval）
            planned_start = interval

            if prevent_overlap:
                # 获取该资源的最近释放时间，默认0（未被占用）
                resource_free_time = self.resource_last_end.get(resource, 0)
                # 若计划启动时间早于资源释放时间，调整到资源释放后启动
                actual_start = max(planned_start, resource_free_time)
                actual_end = actual_start + duration
                
                # 更新资源的最近结束时间
                self.resource_last_end[resource] = actual_end
                self.scheduled.append((job_id, resource, actual_start, actual_end))
            else:
                # 允许重叠，直接按interval安排
                actual_start = planned_start
                actual_end = actual_start + duration
                self.scheduled.append((job_id, resource, actual_start, actual_end))
                # 若需要后续作业参考资源占用情况，可更新结束时间，否则跳过
                # self.resource_last_end[resource] = max(self.resource_last_end.get(resource, 0), actual_end)

        return self.scheduled

关键细节说明 #

1. 资源占用跟踪：用self.resource_last_end字典记录每个资源的最后占用结束时间，O(1)时间就能判断冲突，比遍历所有已调度作业高效得多。
2. 轮询调度机制：通过retry_queue循环处理作业，每次调度一个作业后，资源状态可能变化，之前冲突的作业现在可能可以安排，确保所有作业最终都能被处理。
3. Interval灵活处理：当prevent_resource_overlap=True时，用max(planned_start, resource_free_time)确定实际启动时间，直接覆盖原interval限制，优先保证资源不冲突。
4. 扩展性：可以轻松扩展这个类，比如替换初始时间为真实时间戳、加入作业优先级、自定义冲突解决策略等。

调用示例 #

# 定义作业列表
jobs = [
    {'id': 'job1', 'resource': 'res1', 'duration': 5, 'interval': 0},
    {'id': 'job2', 'resource': 'res1', 'duration': 3, 'interval': 2},
    {'id': 'job3', 'resource': 'res2', 'duration': 4, 'interval': 1},
    {'id': 'job4', 'resource': 'res1', 'duration': 2, 'interval': 3, 'prevent_resource_overlap': False}
]

scheduler = JobScheduler()
result = scheduler.schedule_jobs(jobs)
for item in result:
    print(f"作业{item[0]} 资源{item[1]}: 启动时间{item[2]}，结束时间{item[3]}")

运行后输出符合预期：

• job1占用res1从0到5
• job2因防重叠被安排在5到8（超出原interval=2）
• job4允许重叠，按interval=3安排在3到5，与job1重叠
• job3占用res2从1到5

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