基于simpy的多资源同时请求但时长不同的工厂制造模拟问题

核心思路 #

针对动态长度的生产流程（未知事件数量、资源依赖），采用串行阶段+并行任务的分层处理逻辑：

• 把每个产品的生产流程拆分为多个串行阶段，每个阶段内可以包含多个并行生产任务
• 对每个并行任务单独请求对应机器资源、执行对应时长的生产，等待该阶段所有并行任务完成后，再进入下一个串行阶段
• 全程动态遍历流程节点，无需预先定义固定数量的事件或资源

代码实现示例 #

1. 定义流程与资源 #

import simpy

# 初始化模拟环境
env = simpy.Environment()

# 模拟的机器资源：每个机器是一个simpy.Resource，容量为1（单台机器一次只能处理一个任务）
machine_resources = {
    '机器1': simpy.Resource(env, capacity=1),
    '机器2': simpy.Resource(env, capacity=1),
    '机器3': simpy.Resource(env, capacity=1),
    '机器5': simpy.Resource(env, capacity=1)
}

# 待生产产品的流程列表：每个产品对应一组串行阶段，每个阶段是并行任务集合
product_list = [
    # 产品1：串行阶段依次为「机器1单独生产」→「机器2+机器3并行生产」→「机器5单独生产」
    {
        'name': '产品1',
        'process': [
            [{'machine': '机器1', 'duration': 2}],
            [{'machine': '机器2', 'duration': 3}, {'machine': '机器3', 'duration': 5}],
            [{'machine': '机器5', 'duration': 4}]
        ]
    },
    # 产品2：串行阶段依次为「机器2」→「机器3」→「机器5」
    {
        'name': '产品2',
        'process': [
            [{'machine': '机器2', 'duration': 2}],
            [{'machine': '机器3', 'duration': 3}],
            [{'machine': '机器5', 'duration': 4}]
        ]
    }
]

2. 单个产品的生产进程函数 #

def produce_product(env, product):
    print(f'{env.now:.1f}: 开始生产{product["name"]}')
    for stage_idx, parallel_tasks in enumerate(product['process']):
        print(f'{env.now:.1f}: {product["name"]}进入第{stage_idx+1}阶段')
        # 为每个并行任务创建子进程
        task_events = []
        for task in parallel_tasks:
            machine = machine_resources[task['machine']]
            duration = task['duration']
            # 定义单个任务的执行逻辑
            def run_task(env, machine, duration, task_desc):
                with machine.request() as req:
                    yield req
                    print(f'{env.now:.1f}: 开始执行{task_desc}，时长{duration}')
                    yield env.timeout(duration)
                    print(f'{env.now:.1f}: 完成{task_desc}')
            # 启动任务进程并收集事件
            task_desc = f'{product["name"]}的{task["machine"]}任务'
            task_event = env.process(run_task(env, machine, duration, task_desc))
            task_events.append(task_event)
        # 等待当前阶段所有并行任务完成
        yield env.all_of(task_events)
    print(f'{env.now:.1f}: {product["name"]}生产完成')

3. 启动模拟 #

# 为每个产品启动生产进程
for product in product_list:
    env.process(produce_product(env, product))

# 运行模拟
env.run()

关键细节说明 #

• 动态流程适配：通过遍历product['process']的每个阶段和任务，完全适配未知长度的流程，无需预先硬编码事件
• 并行任务的资源高效利用：每个并行任务独立请求资源、执行对应时长，用env.all_of()等待所有任务完成，不会出现强制取最大时长导致的资源闲置问题
• 资源冲突处理：依赖SimPy原生的Resource请求机制，自动处理机器的抢占与等待逻辑，确保同一机器不会被多个任务同时占用

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