如何在SimPy中建模带工序依赖的车间调度系统？

需求概述 #

• 工序需在前序工序完成后解锁，例如工序1、2可立即启动，工序3需等工序1完成后才能启动，以此类推
• 系统配备多名操作员，仅能在前序任务完成后分配新任务，无法一次性完成全调度

遇到的问题 #

难以在流程中插入后续工序，因逻辑依赖其他工序的完成状态及实时资源可用性：

• 尝试为下一道工序请求操作员，但实际可用的下道工序可能与预期不符，导致系统故障
• 尝试请求所有操作员，使用可用的那个并取消其他请求，但导致仿真提前结束
• 尝试创建自定义事件触发调度，但代码仅能调度第一个工序，完成后便停止运行，代码结构如下：

operation_finished = env.event()

while True:
    """ find what operations and operators are available"""
    available_operations = check_available_operations()
    operators_available = find_any_operator_available()

    """ let the scheduler decide if it wants to schedule a next operation. For now,
 it always does, so I expect these to be null only when the sequence is finished"""
    next_operation = schedule_operation(available_operations, operators_available)
    next_operator = schedule_operator(available_operations, operators_available)

    """ launch the operation with the operator that we know is available.
     This operator will succeed operation_finished once the operation is finished"""
    env.process(operator_process(env, operation))

    """ If the scheduler did not make a decision or no machine was available,
 wait for the next operator to finish"""
    if len(next_operator) == 0:
        yield operation_finished
        print('Whatever is here will never print')

问题分析 #

这段代码的核心问题在于：

1. operation_finished是一次性事件，一旦被触发就会永久处于完成状态，后续yield它会直接返回，无法等待新操作完成
2. 启动任务时传入operation而非选中的next_operation，存在任务匹配错误
3. 有可用操作员时未等待任务完成就进入下一次循环，易引发资源冲突或逻辑混乱

解决方案 #

1. 重构调度核心逻辑 #

改用动态事件跟踪机制，让调度器在每次任务完成后重新扫描资源与可解锁工序：

def scheduler(env, operators, operations):
    while True:
        # 筛选已解锁且未分配的工序
        available_ops = [op for op in operations if op.is_unlocked() and not op.is_assigned()]
        # 筛选空闲操作员
        free_operators = [op for op in operators if op.is_idle()]

        # 执行调度分配（可根据业务规则调整匹配逻辑）
        assigned = False
        for op in available_ops:
            if free_operators:
                operator = free_operators.pop()
                env.process(operator_process(env, operator, op))
                op.mark_assigned()
                assigned = True
                break  # 单批次分配一个任务，可按需修改为批量分配

        # 无任务可分配时，等待任意正在运行的任务完成后重新调度
        if not assigned:
            # 收集所有运行中任务的完成事件
            running_events = [op.finish_event for op in operations if op.is_running()]
            if not running_events:
                # 所有任务完成，退出调度循环
                break
            # 等待任意一个任务完成
            yield env.any_of(running_events)

2. 工序状态与依赖管理 #

为工序添加状态跟踪与依赖解锁逻辑，确保前序完成后自动解锁后续工序：

class Operation:
    def __init__(self, op_id, duration, predecessors=None):
        self.id = op_id
        self.duration = duration
        self.predecessors = predecessors or []
        self._assigned = False
        self._running = False
        self._finished = False
        self.finish_event = env.event()

    def is_unlocked(self):
        # 所有前置工序完成则解锁当前工序
        return all(p._finished for p in self.predecessors)

    def is_assigned(self):
        return self._assigned

    def mark_assigned(self):
        self._assigned = True
        self._running = True

    def is_running(self):
        return self._running

    def mark_finished(self):
        self._running = False
        self._finished = True
        self.finish_event.succeed()

    def unlock_successors(self):
        # 可在此实现通知后续工序的逻辑，或由调度器自动扫描处理
        pass

3. 操作员任务执行逻辑 #

def operator_process(env, operator, operation):
    operator.mark_busy()
    # 模拟工序执行时长
    yield env.timeout(operation.duration)
    operator.mark_idle()
    operation.mark_finished()
    # 解锁当前工序的后续任务
    operation.unlock_successors()

关键改进点 #

• 每次调度循环重新扫描资源与工序，确保依赖变化被及时感知
• 用每个工序独立的finish_event跟踪任务完成，避免单一事件的一次性问题
• 明确工序的解锁、分配、运行、完成状态，逻辑边界清晰
• 无任务可分配时等待任意任务完成，不会提前终止仿真

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