简介:一份面向Python开发者与系统仿真初学者的完整技术文档,以SimPy离散事件建模为主线,帮助读者从零掌握复杂系统仿真的核心方法。文档内容系统完整,先从离散事件系统的基本概念讲起,再深入讲解SimPy的环境、进程、事件、资源等核心组件,并提供了医院急诊室、交通路口、供应链系统三大真实场景的完整仿真案例实现,每个案例都包含问题描述、模型假设、代码实现与结果分析。同时,还涵盖了仿真结果的数据收集与可视化分析、自定义事件与进程、资源优先级管理、多环境并行仿真等进阶应用技巧。资源为1个PDF文件,共32页,大小4.98MB,支持目录章节跳转与阅读器大纲定位,排版清晰完整。已有207人学习使用,适合需要快速上手SimPy进行复杂系统建模与仿真的技术开发人员和高校学生。
1. 为什么拿 SimPy 做离散事件建模:先解决「资源排队」这类算不清的账
做排班优化、产线扩容或者急诊室流程改造的时候,最头疼的问题不是找不到方案,而是没人说得清方案投下去到底有没有用。拿医院急诊室举例,加一个医生到底能让平均等待时间降多少?新增一台设备值不值?这类问题靠 Excel 手工算算不动,靠写多线程模拟要处理大量并发和时间管理,成本又太高。SimPy 正是为这种场景设计的:它基于 Python 的生成器机制,用事件驱动的思路模拟「谁在什么时刻做了什么、占用了什么资源、等了多少时间」。这篇笔记会从核心概念、建模范式、三个实战案例到高频踩坑逐个拆,让你拿到就能照着搭自己的仿真模型。
2. 核心组件与基本建模范式:环境、进程、事件怎么配合
2.1 环境与进程:仿真世界的「时钟」和「演员」
先建立两个最基本的认知:环境(Environment)管时间,进程(Process)管行为。Environment 内部维护一个事件队列和当前仿真时刻(env.now),所有的事件都按时间顺序排列,环境只负责按次序把事件拿出来执行。进程则是写在生成器函数里的行为逻辑,用yield暂停和恢复执行。
这里的关键是理解yield的语义:进程遇到yield env.timeout(n)时,相当于说「我先睡 n 个时间单位,到点再叫我」。环境把这条等待记录挂进事件队列,然后把执行权交给其他进程。SimPy 是协作式调度,不是多线程抢占,所有进程都在同一个 Python 线程里依次切换,不存在共享内存竞争问题。
import simpy def car(env): while True: # 停车阶段 print(f'开始停车,当前时间: {env.now}') yield env.timeout(5) # 行驶阶段 print(f'开始行驶,当前时间: {env.now}') yield env.timeout(2) env = simpy.Environment() env.process(car(env)) env.run(until=15)逻辑说明:car是个无限循环生成器,env.timeout(5)让进程暂停 5 个时间单位,env.timeout(2)再暂停 2 个。env.run(until=15)的意思是让仿真时钟一直推进到 15 为止,进程会循环执行停车和行驶的动作。参数until可以设成整数也可以是datetime.timedelta对象,后者常用于真实日期时间的仿真。跑这段代码会看到时间点 0、5、7、12、14 交替出现,这就是环境按事件时间戳调度进程的直观效果。
一个重要认知:进程的执行顺序不取决于env.process()的调用顺序,而取决于事件被唤醒的时间顺序。哪怕你先注册了后到的进程,只要它的唤醒时间靠后,它就必须等前面的时间事件先处理完。理解这一点对调试多进程交互非常重要。
2.2 事件机制:yield 暂停与恢复的底层逻辑
事件(Event)是 SimPy 的基本调度单位,所有能让进程暂停的结构本质上都是 Event。最常用的是env.timeout(delay),它在env.now + delay时刻触发。还有event.succeed()手动触发的同步事件,以及simpy.Resource的资源请求事件。进程通过yield event来「等待」,事件一旦触发,进程从yield处恢复继续往下执行。
import simpy def event_example(env): print(f"当前时间: {env.now}") # 创建一个超时事件 event = env.timeout(5) # 等待事件触发 yield event print(f"超时事件触发,当前时间: {env.now}") env = simpy.Environment() env.process(event_example(env)) env.run()逻辑说明:env.timeout(5)拿到的是一个事件对象,yield event把进程挂起直到事件触发。当仿真环境处理到这个超时事件时,进程被唤醒并继续执行后续代码。注意env.run()不带参数时会一直运行到所有活跃进程全部终止,对这个例子来说事件处理完进程即结束,仿真自然停止。
参数层面,timeout的延迟单位完全由你的模型自定义。你可以约定 1 个单位 = 1 分钟,也可以 = 1 小时,SimPy 本身不做单位换算。这点从建模第一天就必须明确,不然后期分析和真实系统对比数据时容易出偏差。我一般会在脚本头部写清楚时间单位的注释,并统一用同一个单位常量。
3. Resource 与随机建模:排队系统的核心套路
3.1 Resource:代表服务台、医生、工位这些有限容量资源
离散事件仿真里最常遇到的场景就是排队:顾客到收银台结账、患者到诊室看病、工件到机床加工。这类系统的共同点是资源有限、需求随机。SimPy 的 Resource 组件正是为此设计的,它内部维护一个容量计数和等待队列,capacity参数决定同时可服务的数量。
import simpy import random # 平均顾客到达间隔时间 ARRIVAL_INTERVAL = 5 # 平均服务时间 SERVICE_TIME = 3 def customer(env, name, counter): print(f'{name} 于 {env.now} 到达超市') with counter.request() as req: yield req print(f'{name} 于 {env.now} 开始接受服务') service_time = random.expovariate(1 / SERVICE_TIME) yield env.timeout(service_time) print(f'{name} 于 {env.now} 完成服务离开超市') def customer_arrivals(env, counter): i = 0 while True: yield env.timeout(random.expovariate(1 / ARRIVAL_INTERVAL)) i += 1 env.process(customer(env, f'顾客 {i}', counter)) env = simpy.Environment() counter = simpy.Resource(env, capacity=1) env.process(customer_arrivals(env, counter)) env.run(until=20)逻辑说明:customer进程用with counter.request() as req申请收银台资源,申请成功后才进入服务状态。with块的好处是无论正常结束还是中途异常,资源都会自动释放,这是 SimPy 最推荐的写法。customer_arrivals进程用指数分布生成随机到达间隔,每到一个顾客就调env.process()创建新进程。capacity=1表示只有一个收银台,如果改成 3 就意味着三个收银台并行服务,这是实验参数里最值得调优的变量。
这里有个容易被忽略的细节:request()返回的请求对象在进入with块时可能不会立即满足——如果当前资源被占满,进程在这个yield处挂起,进入资源的等待队列。资源空闲时按队列顺序唤醒对应的进程。SimPy 默认是 FIFO 队列,PriorityResource则支持按优先级插队,这在后面的高级章节会展开。
3.2 随机分布与仿真参数:为什么用指数分布来模拟到达
大多数排队系统里,顾客到达过程不是均匀间隔的。进店的客人总是一阵一阵来,有时连续两三个人,有时五分钟没人。指数分布模拟这种「无记忆性」的随机到达最合适——它对应泊松过程的到达间隔,每个时间点到达概率恒定,与已经等了多少时间无关。代码里random.expovariate(1 / ARRIVAL_INTERVAL)的lambd参数是到达率,1 / 5表示平均每 5 分钟来一个。
import simpy import random import numpy as np import matplotlib.pyplot as plt ARRIVAL_INTERVAL = 5 SERVICE_TIME = 3 SIMULATION_TIME = 100 waiting_times = [] def customer(env, name, server): arrival_time = env.now with server.request() as req: yield req waiting_time = env.now - arrival_time waiting_times.append(waiting_time) service_duration = random.expovariate(1 / SERVICE_TIME) yield env.timeout(service_duration) def customer_arrivals(env, server): i = 0 while True: yield env.timeout(random.expovariate(1 / ARRIVAL_INTERVAL)) i += 1 env.process(customer(env, f'顾客 {i}', server)) env = simpy.Environment() server = simpy.Resource(env, capacity=1) env.process(customer_arrivals(env, server)) env.run(until=SIMULATION_TIME) # 用 NumPy 计算平均等待时间 average_waiting_time = np.mean(waiting_times) print(f'平均等待时间: {average_waiting_time}') # 用 Matplotlib 绘制等待时间直方图 plt.hist(waiting_times, bins=20, edgecolor='k') plt.xlabel('等待时间') plt.ylabel('顾客数量') plt.title('顾客等待时间直方图') plt.show()逻辑说明:这段代码在上一例基础上增加了数据收集——waiting_times列表记录每个顾客从到达资源请求到真正获得服务的等待时长。仿真结束后用 NumPy 算平均值,用 Matplotlib 画直方图看分布形态。参数until=100把仿真时长拉长到 100 个时间单位,为了让采样量足够大,统计结果才有代表性。
参数选择上要注意:SERVICE_TIME = 3是平均服务时长,指数分布的方差等于均值平方,所以波动很大。真实场景中如果想模拟更稳定的服务时间,可以用正态分布(random.gauss)或三角分布,需要根据实测数据的形态来定。做敏感性分析时一般会把这几个参数设为变量,跑多组取对比。
4. 三个复杂系统案例拆解:从模型假设到可运行代码
4.1 医院急诊室:分诊护士和医生的两级排队
急诊室的典型场景是患者到达后先分诊、再就诊。分诊护士是第一个瓶颈,医生资源是第二个瓶颈。建模的核心是先画流程、再定资源容量和到达分布。模型假设:患者平均每 6 分钟到一位,分诊服务时间服从均值为 2 的指数分布,医生接诊时间均值为 10 分钟,共有 2 名医生。
import simpy import random ARRIVAL_INTERVAL = 6 TRIAGE_TIME = 2 DOCTOR_TIME = 10 NUM_DOCTORS = 2 SIM_TIME = 480 # 模拟一个 8 小时班次 def patient(env, name, triage_nurse, doctor): arrival_time = env.now # 分诊阶段 with triage_nurse.request() as req: yield req yield env.timeout(random.expovariate(1 / TRIAGE_TIME)) # 候诊阶段:等医生 with doctor.request() as req: yield req yield env.timeout(random.expovariate(1 / DOCTOR_TIME)) total_time = env.now - arrival_time print(f'{name} 总耗时 {total_time:.1f}') def patient_arrivals(env, triage_nurse, doctor): i = 0 while True: yield env.timeout(random.expovariate(1 / ARRIVAL_INTERVAL)) i += 1 env.process(patient(env, f'患者 {i}', triage_nurse, doctor)) env = simpy.Environment() triage_nurse = simpy.Resource(env, capacity=1) doctor = simpy.Resource(env, capacity=NUM_DOCTORS) env.process(patient_arrivals(env, triage_nurse, doctor)) env.run(until=SIM_TIME)逻辑说明:患者进程先请求分诊护士资源,完成后立刻请求医生资源。两级资源串行构成了两级队列。capacity=NUM_DOCTORS的调参意义在于:把NUM_DOCTORS从 2 改成 3 再跑一遍,对比总耗时的变化,就能判断新增医生的边际收益。常见做法是记录每个患者的total_time到列表里,最后画直方图看长尾在哪。这个模型里没有区分患者危重程度,实际项目建议把患者分类为轻症和重症,用PriorityResource让重症优先插队。
4.2 交通路口:信号灯周期和车辆排队等待
交通路口仿真的核心是信号灯状态切换对车辆到达/离开节奏的影响。用一个 Simulation 模拟主干道单路口:车辆随机到达停车线,红灯亮时排队等待,绿灯亮时依次通过。信号灯周期固定:红灯 30 秒、绿灯 40 秒,循环切换。
import simpy RED_TIME = 30 GREEN_TIME = 40 ARRIVE_INTERVAL = 3 SIM_TIME = 600 def traffic_light(env, green_event): while True: # 红灯阶段:红灯不单独建模,绿色事件未触发即禁止通行 yield env.timeout(RED_TIME) # 绿灯亮:触发事件唤醒所有等待车辆 green_event.succeed() green_event = env.event() yield env.timeout(GREEN_TIME) green_event = env.event() def car(env, name, green_event): print(f'{name} 在 {env.now} 到达路口') yield green_event print(f'{name} 在 {env.now} 通过路口') def car_arrivals(env, green_event): i = 0 while True: yield env.timeout(ARRIVE_INTERVAL) i += 1 env.process(car(env, f'车辆 {i}', green_event)) env = simpy.Environment() green_event = env.event() env.process(traffic_light(env, green_event)) env.process(car_arrivals(env, green_event)) env.run(until=SIM_TIME)逻辑说明:green_event是手动管理的事件对象。红灯期间到达的车辆执行yield green_event挂起。绿灯亮时green_event.succeed()会唤醒所有等待车辆,然后重建一个新事件对象供下一轮使用。注意succeed()只能触发一次,所以必须在每个周期都创建新的事件实例。参数ARRIVE_INTERVAL和RED_TIME、GREEN_TIME的比例直接决定拥堵程度,可调的实验杠杆是绿灯时长分配。
这个模型做了简化:所有车辆到达即排队,不考虑通过时间,也不区分左转直行。如果要模拟多车道和多相位,建议把信号灯状态机拆成独立的 Process,每个相位用env.timeout推进,相位切换时更新通行事件。逻辑虽然复杂,但核心思路还是「挂起等待 + 事件唤醒」。
4.3 供应链系统:库存消耗、补货触发与安全库存验证
供应链仿真的典型问题是确定再订货点和订货量:库存太低容易缺货,太高占用资金。一个简化的仓库模型:需求随机到达,每单消耗 1 件库存,库存低于再订货点时触发补货,补货提前期固定 5 天,记录缺货次数和平均库存水位。
import simpy import random INITIAL_STOCK = 20 REORDER_POINT = 10 ORDER_QTY = 30 LEAD_TIME = 5 DEMAND_INTERVAL = 2 SIM_TIME = 100 stock = INITIAL_STOCK reorder_in_transit = 0 stock_out_count = 0 stock_levels = [] def demand_process(env): global stock, reorder_in_transit, stock_out_count while True: yield env.timeout(random.expovariate(1 / DEMAND_INTERVAL)) stock_levels.append(stock) if stock > 0: stock -= 1 else: stock_out_count += 1 # 达到再订货点且没有在途订单,触发补货 if stock <= REORDER_POINT and reorder_in_transit == 0: reorder_in_transit = 1 env.process(replenishment(env)) def replenishment(env): global stock, reorder_in_transit yield env.timeout(LEAD_TIME) stock += ORDER_QTY reorder_in_transit = 0 env = simpy.Environment() env.process(demand_process(env)) env.run(until=SIM_TIME) print(f'缺货次数: {stock_out_count}') print(f'平均库存: {sum(stock_levels) / len(stock_levels):.1f}')逻辑说明:用全局变量stock维护库存水位。reorder_in_transit是个布尔标志,防止重复下单。stock <= REORDER_POINT触发补货逻辑,LEAD_TIME是下单到入库的延迟,用来模拟真实供应链的滞后。每次需求到达后记录库存快照,仿真结束计算平均库存。调整REORDER_POINT和ORDER_QTY这两个参数观察缺货率变化,就是这个模型的核心实验。
这里有个容易被忽略的细节:库存统计的时间加权问题。每步固定间隔采样的stock_levels是离散快照,如果需求不是均匀到达,均值会和真实时间加权平均有一些偏差。更精确的做法是在每次库存变化时记录变化时刻和变化值,用时间加权公式计算平均库存。小规模模型用快照法问题不大,跑敏感性分析时建议换成事件驱动的统计方式。
5. 仿真常见问题与避坑记录:五个翻车现场
5.1 进程死循环:env.run() 永远跑不完
现象:设置了until=100,程序却像卡死一样一直不结束,CPU 占用率拉满。
原因:进程内的while True循环里有yield env.timeout(0)或直接yield一个已触发的事件,事件循环不停地把零时延事件塞回队列,仿真时间始终不前进。这在场景里最常出现在那些忘了加超时时间的分支逻辑上。
解决:检查所有yield语句,确认每次循环至少有一个正时长的env.timeout()。排查时可以临时在进程开头加计数器,超过某个阈值就抛异常退出。我一般会在每个while True循环的入口打一行日志,跑到哪个分支立刻能看出来。
5.2 随机结果不可复现:两次跑同一模型结果差很大
现象:改了一个参数后重跑,发现连改之前的结果都对不上了,很难判断效果到底是参数引起的还是随机波动引起的。
原因:没有固定随机种子。random.expovariate每次调用都用系统熵源生成结果,两次运行从第一分钟起就分道扬镳了。
解决:在脚本开头调用random.seed(42)。生产级实验应该把种子作为命令行参数,每轮实验记录种子值以便回归。这样做敏感性分析时,每个参数组合都用相同的随机数序列,对比才有意义。
5.3 with 块缩进错误:统计数据被提前记录了
现象:等待时间的列表里出现很多 0 值,明显不符合直觉。
原因:把统计等待时间的append语句写在了with server.request()之前,进程还在排队呢就已经把等待时间记录成 0 了。
解决:记录逻辑必须放在yield req之后——只有yield返回了,资源才算真正拿到,此刻env.now - arrival_time才是真实等待时长。拿不准的时候把打印语句放在不同位置跑一次短仿真,日志顺序一目了然。
5.4 事件对象复用导致状态丢失
现象:手动管理的事件在第二个周期失效,进程在应该被唤醒时继续挂起。
原因:event.succeed()调用后事件就进入触发状态,不能用于下一轮。没有为每个周期创建新的事件实例,导致后续车辆永远等一个已经触发过的事件。
解决:在触发后立刻重新赋值一个新事件对象,像交通灯示例里那样green_event = env.event()。凡是手动管理的周期事件,都要养成「触发即重建」的习惯。
5.5 时间单位不统一:参数看起来对但结果离谱
现象:改成真实数据后仿真结果比实际情况大几百倍,或者小几百倍。
原因:有人把到达间隔写成「5 分钟」,服务时间写成「3 秒」,直接混用没换算,仿真输出完全不可解释。
解决:在脚本头部定义明确的单位常量并写注释,例如TIME_UNIT = 1 # 1 单位 = 1 分钟,所有参数都基于这个单位换算。如果项目需要跨系统对比,建议用datetime.timedelta作为底层表示,SimPy 官方支持,只是速度慢一些。
6. 最后一块拼图:事件组合、优先级资源与可视化验证
6.1 事件组合与条件等待:从「等一个」到「等一组」
SimPy 支持用&和|组合事件。event1 & event2表示两个事件都触发后才会唤醒等待进程;event1 | event2表示任一事件触发即可唤醒。这在建模「等物料到位同时等设备空闲」这类并联条件时非常实用。
import simpy def machine_repair(env): # 物料到达和维修工空闲两个条件同时满足才开工 parts_ready = env.timeout(4) worker_free = env.timeout(6) yield parts_ready & worker_free print(f'维修开始,当前时间: {env.now}') env = simpy.Environment() env.process(machine_repair(env)) env.run()逻辑说明:parts_ready在时刻 4 触发,worker_free在时刻 6 触发,&组合事件要等两者都完成,所以env.now输出是 6。如果用|就会变成 4——谁先到都算满足。注意组合事件的触发时间只取决于较快或较慢的那个边界,不能用来表达「还需要再等一段固定时间」的语义,那种情况还是得用env.timeout。
6.2 优先级资源与可视化验证
PriorityResource是Resource的子类,request(priority=1)可以指定优先级。数字越小优先级越高,高优先级请求会排在低优先级前面抢占资源。医院分诊里给危重患者高优先级、供应链里给关键客户订单提前插队,都是典型用法。
from simpy.resources.resource import PriorityResource resource = PriorityResource(env, capacity=1) # 低优先级请求先到,高优先级后到,但高优先级先获得资源 env.process(low_priority_process(env, resource)) env.process(high_priority_process(env, resource))可视化方面,记住一个原则:先看分布再看均值。plt.hist(waiting_times, bins=20)看长尾,plt.boxplot(waiting_times)看离群点,plt.plot(env.now_list, stock_levels)看随时间的变化趋势。多组参数对比时把多条折线画在同一张图里,一眼就能看出瓶颈在哪组参数下消失。从那以后我每次跑仿真都强制走一遍固定种子、时间单位注释、事件实例重建检查这三步,再做任何参数实验心里都有底。希望帮到你。
本文还有配套的精品资源,点击获取