如何使用SimPy模拟需等待坐满乘客的公交车？

核心问题分析 #

你的代码存在几个关键问题：

• 队列操作错误：passengers_queue(env.now) 是把列表当函数调用，正确应该用 append 方法添加乘客
• 同步循环误用：SimPy是事件驱动模型，不能用普通while循环死等乘客，会阻塞整个模拟进程
• 乘客生成逻辑混乱：每次公交车到站就启动新的passenger_arrival进程，会导致乘客重复生成
• 变量作用域缺失：total_amount_of_spots、passengers_queue等变量未正确定义或传递

修复后的完整代码 #

import simpy

# 全局配置参数
TOTAL_SPOTS = 5  # 公交车总座位数
BUS_INTERVAL = 20  # 公交车到站间隔时间
PASSENGER_INTERVAL = 5  # 乘客到达间隔时间

def arrival_bus(env, bus, passengers_queue):
    print("-----------------------------------")
    print(f"<<<<<<<<<<<< BUS ARRIVES AT STATION (time: {env.now})")
    
    bus['free_spots'] = TOTAL_SPOTS
    print(f"CURRENT PASSENGERS IN STATION: {len(passengers_queue)}")

    # 先让车站现有乘客上车
    if passengers_queue:
        board_num = min(len(passengers_queue), bus['free_spots'])
        bus['free_spots'] -= board_num
        # 移除已上车的乘客
        del passengers_queue[:board_num]
        print(f"BOARDED {TOTAL_SPOTS - bus['free_spots']} PASSENGERS")

    # 等待剩余空位被填满
    if bus['free_spots'] > 0:
        print(f"WAITING FOR {bus['free_spots']} MORE PASSENGERS...")
        # 用事件驱动方式等待，避免阻塞模拟
        while len(passengers_queue) < bus['free_spots']:
            yield env.timeout(0)
    
    # 公交车坐满发车
    print(f">>>>>>>>>>>> BUS DEPARTS (time: {env.now}), FULLY OCCUPIED")
    # 模拟公交车行驶时间（可选，这里设为10单位时间）
    yield env.timeout(10)

def passenger_arrival(env, passengers_queue):
    while True:
        yield env.timeout(PASSENGER_INTERVAL)
        passenger_id = env.now  # 用当前时间作为乘客唯一标识
        passengers_queue.append(passenger_id)
        print(f"PASSENGER {passenger_id} ARRIVED (time: {env.now})")
        print(f"PASSENGERS IN STATION: {len(passengers_queue)}")

def bus_movement(env, passengers_queue):
    # 用字典表示公交车状态
    bus = {'free_spots': TOTAL_SPOTS}
    while True:
        yield env.timeout(BUS_INTERVAL)
        # 等待当前公交车完成载客流程，再调度下一班
        yield env.process(arrival_bus(env, bus, passengers_queue))

# 启动模拟
if __name__ == '__main__':
    env = simpy.Environment()
    passengers_queue = []
    # 启动全局乘客生成进程
    env.process(passenger_arrival(env, passengers_queue))
    # 启动公交车调度进程
    env.process(bus_movement(env, passengers_queue))
    # 运行模拟100单位时间
    env.run(until=100)

关键修改说明 #

• 队列操作修正：改用append添加乘客，del移除已上车乘客，符合列表操作逻辑
• 事件驱动等待：将死循环替换为yield env.timeout(0)，让SimPy有机会处理其他事件（比如乘客到达），避免模拟阻塞
• 乘客生成全局化：仅启动一次passenger_arrival进程，持续生成乘客，避免重复创建进程导致的异常
• 进程同步：bus_movement中用yield env.process(...)等待当前公交车完成载客再调度下一班，保证流程顺序
• 变量传递优化：将passengers_queue作为参数传递，避免全局变量的潜在冲突
• 时间可视化：添加模拟时间戳，方便跟踪事件发生顺序

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