当前位置：首页 > news >正文

SimPy经典案例分析

news 2025/11/2 10:18:44

SimPy 是 Python 中用于离散事件仿真（DES）的经典库，广泛应用于排队系统、资源管理、生产调度、交通仿真等领域。以下通过几个经典案例，解析 SimPy 的核心思想（进程、资源、事件调度）及实际建模方法。

在这里插入图片描述

案例 1：银行单服务台排队系统

问题背景：模拟银行单柜员服务场景，顾客随机到达，等待柜员服务，服务完成后离开。需统计平均等待时间、队列长度等指标。

模型核心要素

顾客进程：生成顾客到达事件，请求服务，完成后离开。
服务台资源：单个柜员（simpy.Resource），处理顾客服务请求。
统计指标：顾客等待时间（到达时间→开始服务时间）、队列长度变化。

关键代码

import simpy
import randomdef customer(env, name, counter):# 顾客到达，记录到达时间arrival_time = env.nowprint(f"{arrival_time:.2f}: {name} 到达银行")# 请求服务台（等待直到有空闲柜员）with counter.request() as req:yield req  # 等待资源可用# 开始服务，计算等待时间wait_time = env.now - arrival_timeprint(f"{env.now:.2f}: {name} 开始服务，等待了 {wait_time:.2f}")# 服务时间（指数分布，均值 5）service_time = random.expovariate(1/5)yield env.timeout(service_time)print(f"{env.now:.2f}: {name} 离开银行")def setup(env, num_counters, num_customers):# 创建服务台资源（单柜员）counter = simpy.Resource(env, capacity=num_counters)# 生成顾客到达事件（泊松过程，均值 10 到达间隔）for i in range(num_customers):inter_arrival = random.expovariate(1/10)env.process(customer(env, f"顾客{i}", counter))yield env.timeout(inter_arrival)# 运行仿真：1 个柜员，100 个顾客
env = simpy.Environment()
env.process(setup(env, num_counters=1, num_customers=100))
env.run()# 统计：可扩展记录等待时间列表，计算均值

结果分析

关键指标：通过记录每个顾客的 wait_time，可计算平均等待时间（反映服务质量）。
敏感性测试：调整柜员数量（num_counters）或服务时间分布，观察队列长度和等待时间的变化（符合排队论 M/M/1 模型）。

在这里插入图片描述

案例 2：超市多收银台与动态队列选择

问题背景：超市有多个收银台，顾客选择最短队列结账。需模拟顾客动态选队行为，并分析不同收银台效率对整体吞吐量的影响。

模型核心改进

多资源竞争：多个收银台（simpy.Resource 实例），顾客主动选择队列最短的收银台。
动态选队策略：顾客到达时遍历所有收银台，选择当前等待人数最少的队列。

关键代码片段

def customer(env, name, cashiers):arrival_time = env.nowprint(f"{arrival_time:.2f}: {name} 到达超市")# 动态选择最短队列的收银台selected_cashier = min(cashiers, key=lambda c: len(c.users))queue_length = len(selected_cashier.users)print(f"{arrival_time:.2f}: {name} 选择 {selected_cashier.capacity}-号收银台，当前队列 {queue_length} 人")with selected_cashier.request() as req:yield reqwait_time = env.now - arrival_timeprint(f"{env.now:.2f}: {name} 开始结账，等待 {wait_time:.2f}")service_time = random.uniform(1, 5)  # 结账时间均匀分布yield env.timeout(service_time)print(f"{env.now:.2f}: {name} 结账完成")def setup(env, num_cashiers, num_customers):cashiers = [simpy.Resource(env, capacity=1) for _ in range(num_cashiers)]for i in range(num_customers):inter_arrival = random.expovariate(1/3)  # 平均 3 分钟到达一个顾客env.process(customer(env, f"顾客{i}", cashiers))yield env.timeout(inter_arrival)# 运行：4 个收银台的超市，200 个顾客
env = simpy.Environment()
env.process(setup(env, num_cashiers=4, num_customers=200))
env.run()

扩展分析

队列分配策略：可对比“最短队列”与“随机选择”“固定队列”的效率差异。
资源利用率：统计每个收银台的服务时间占比（len(c.users)/总时间），优化人员配置。

在这里插入图片描述

案例 3：计算机系统 CPU 调度与进程竞争

问题背景：模拟操作系统 CPU 调度，多个进程竞争 CPU 资源，进程可能因 I/O 阻塞，释放 CPU 后其他进程抢占。

模型核心要素

CPU 资源：单核 CPU（simpy.Resource(capacity=1)）。
进程行为：每个进程包含 CPU 计算（占用 CPU）和 I/O 等待（释放 CPU，延迟后恢复）。

关键代码

def process(env, name, cpu):# 进程执行阶段 1：计算（占用 CPU）with cpu.request() as req:yield reqcompute_time = random.uniform(0.5, 2)print(f"{env.now:.2f}: {name} 开始计算，耗时 {compute_time:.2f}")yield env.timeout(compute_time)# I/O 等待（释放 CPU，不占用资源）io_time = random.uniform(1, 3)print(f"{env.now:.2f}: {name} 开始 I/O，耗时 {io_time:.2f}")yield env.timeout(io_time)# 进程执行阶段 2：再次计算with cpu.request() as req:yield reqcompute_time = random.uniform(0.5, 2)print(f"{env.now:.2f}: {name} 恢复计算，耗时 {compute_time:.2f}")yield env.timeout(compute_time)print(f"{env.now:.2f}: {name} 完成")def setup(env, num_processes):cpu = simpy.Resource(env, capacity=1)for i in range(num_processes):env.process(process(env, f"进程{i}", cpu))yield env.timeout(random.expovariate(1/5))  # 进程到达间隔# 运行：5 个进程竞争单核 CPU
env = simpy.Environment()
env.process(setup(env, num_processes=5))
env.run()

关键结论

进程切换：通过 with cpu.request() 上下文管理器，自动释放资源，模拟进程抢占。
吞吐量：统计单位时间内完成的进程数，评估 CPU 调度效率。

在这里插入图片描述

案例 4：交通信号灯控制的十字路口

问题背景：模拟十字路口车辆通行，东西/南北方向交替放行（红绿灯），车辆在红灯时排队，绿灯时按顺序通过。

模型核心设计

时间触发事件：红绿灯周期性切换（如东西绿灯 30s，南北绿灯 30s）。
车辆进程：车辆到达路口，根据当前灯色决定等待或通过。

关键代码逻辑

class TrafficLight:def __init__(self, env):self.env = envself.direction = "east-west"  # 初始方向：东西绿灯self.action = env.process(self.run())def run(self):while True:# 东西绿灯 30sprint(f"{self.env.now:.0f}: 东西方向绿灯")yield self.env.timeout(30)# 切换为南北绿灯 30sself.direction = "north-south"print(f"{self.env.now:.0f}: 切换为南北方向绿灯")yield self.env.timeout(30)self.direction = "east-west"def car(env, name, traffic_light):arrival_time = env.nowdirection = random.choice(["east-west", "north-south"])print(f"{arrival_time:.0f}: {name} 从 {direction} 方向到达路口")while True:if traffic_light.direction == direction:# 绿灯，尝试通过（假设无冲突，直接通过）print(f"{env.now:.0f}: {name} 通过路口")breakelse:# 红灯，等待 1s 后重试（避免忙等待）yield env.timeout(1)print(f"{env.now:.0f}: {name} 等待红灯")# 运行：初始化红绿灯和 20 辆车
env = simpy.Environment()
traffic_light = TrafficLight(env)
for i in range(20):env.process(car(env, f"车辆{i}", traffic_light))yield env.timeout(random.uniform(0, 5))  # 车辆随机到达
env.run(until=100)  # 运行 100 时间单位

扩展方向

冲突检测：加入左转车辆与对向直行车辆的冲突，需更复杂的信号配时。
队列长度限制：路口设置最大排队长度，超过则车辆绕行。

SimPy 经典案例总结

案例类型	核心模型要素	关键 SimPy 功能	应用价值
银行排队	单资源、顾客到达/离开	`Resource.request()`、`timeout`	优化服务台数量，降低等待时间
超市多收银台	多资源竞争、动态队列选择	多 `Resource` 实例、最小队列策略	提升结账效率，减少拥堵
CPU 进程调度	资源抢占、I/O 阻塞	上下文管理器释放资源	评估操作系统调度算法
交通信号灯	时间触发事件、状态切换	`env.process` 周期任务	优化路口通行效率