简介一份面向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(until15)逻辑说明car是个无限循环生成器env.timeout(5)让进程暂停 5 个时间单位env.timeout(2)再暂停 2 个。env.run(until15)的意思是让仿真时钟一直推进到 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, capacity1) env.process(customer_arrivals(env, counter)) env.run(until20)逻辑说明customer进程用with counter.request() as req申请收银台资源申请成功后才进入服务状态。with块的好处是无论正常结束还是中途异常资源都会自动释放这是 SimPy 最推荐的写法。customer_arrivals进程用指数分布生成随机到达间隔每到一个顾客就调env.process()创建新进程。capacity1表示只有一个收银台如果改成 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, capacity1) env.process(customer_arrivals(env, server)) env.run(untilSIMULATION_TIME) # 用 NumPy 计算平均等待时间 average_waiting_time np.mean(waiting_times) print(f平均等待时间: {average_waiting_time}) # 用 Matplotlib 绘制等待时间直方图 plt.hist(waiting_times, bins20, edgecolork) plt.xlabel(等待时间) plt.ylabel(顾客数量) plt.title(顾客等待时间直方图) plt.show()逻辑说明这段代码在上一例基础上增加了数据收集——waiting_times列表记录每个顾客从到达资源请求到真正获得服务的等待时长。仿真结束后用 NumPy 算平均值用 Matplotlib 画直方图看分布形态。参数until100把仿真时长拉长到 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, capacity1) doctor simpy.Resource(env, capacityNUM_DOCTORS) env.process(patient_arrivals(env, triage_nurse, doctor)) env.run(untilSIM_TIME)逻辑说明患者进程先请求分诊护士资源完成后立刻请求医生资源。两级资源串行构成了两级队列。capacityNUM_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(untilSIM_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(untilSIM_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() 永远跑不完现象设置了until100程序却像卡死一样一直不结束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(priority1)可以指定优先级。数字越小优先级越高高优先级请求会排在低优先级前面抢占资源。医院分诊里给危重患者高优先级、供应链里给关键客户订单提前插队都是典型用法。from simpy.resources.resource import PriorityResource resource PriorityResource(env, capacity1) # 低优先级请求先到高优先级后到但高优先级先获得资源 env.process(low_priority_process(env, resource)) env.process(high_priority_process(env, resource))可视化方面记住一个原则先看分布再看均值。plt.hist(waiting_times, bins20)看长尾plt.boxplot(waiting_times)看离群点plt.plot(env.now_list, stock_levels)看随时间的变化趋势。多组参数对比时把多条折线画在同一张图里一眼就能看出瓶颈在哪组参数下消失。从那以后我每次跑仿真都强制走一遍固定种子、时间单位注释、事件实例重建检查这三步再做任何参数实验心里都有底。希望帮到你。本文还有配套的精品资源点击获取 用Python SimPy实现离散事件仿真：从资源排队到系统优化
简介一份面向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(until15)逻辑说明car是个无限循环生成器env.timeout(5)让进程暂停 5 个时间单位env.timeout(2)再暂停 2 个。env.run(until15)的意思是让仿真时钟一直推进到 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, capacity1) env.process(customer_arrivals(env, counter)) env.run(until20)逻辑说明customer进程用with counter.request() as req申请收银台资源申请成功后才进入服务状态。with块的好处是无论正常结束还是中途异常资源都会自动释放这是 SimPy 最推荐的写法。customer_arrivals进程用指数分布生成随机到达间隔每到一个顾客就调env.process()创建新进程。capacity1表示只有一个收银台如果改成 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, capacity1) env.process(customer_arrivals(env, server)) env.run(untilSIMULATION_TIME) # 用 NumPy 计算平均等待时间 average_waiting_time np.mean(waiting_times) print(f平均等待时间: {average_waiting_time}) # 用 Matplotlib 绘制等待时间直方图 plt.hist(waiting_times, bins20, edgecolork) plt.xlabel(等待时间) plt.ylabel(顾客数量) plt.title(顾客等待时间直方图) plt.show()逻辑说明这段代码在上一例基础上增加了数据收集——waiting_times列表记录每个顾客从到达资源请求到真正获得服务的等待时长。仿真结束后用 NumPy 算平均值用 Matplotlib 画直方图看分布形态。参数until100把仿真时长拉长到 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, capacity1) doctor simpy.Resource(env, capacityNUM_DOCTORS) env.process(patient_arrivals(env, triage_nurse, doctor)) env.run(untilSIM_TIME)逻辑说明患者进程先请求分诊护士资源完成后立刻请求医生资源。两级资源串行构成了两级队列。capacityNUM_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(untilSIM_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(untilSIM_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() 永远跑不完现象设置了until100程序却像卡死一样一直不结束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(priority1)可以指定优先级。数字越小优先级越高高优先级请求会排在低优先级前面抢占资源。医院分诊里给危重患者高优先级、供应链里给关键客户订单提前插队都是典型用法。from simpy.resources.resource import PriorityResource resource PriorityResource(env, capacity1) # 低优先级请求先到高优先级后到但高优先级先获得资源 env.process(low_priority_process(env, resource)) env.process(high_priority_process(env, resource))可视化方面记住一个原则先看分布再看均值。plt.hist(waiting_times, bins20)看长尾plt.boxplot(waiting_times)看离群点plt.plot(env.now_list, stock_levels)看随时间的变化趋势。多组参数对比时把多条折线画在同一张图里一眼就能看出瓶颈在哪组参数下消失。从那以后我每次跑仿真都强制走一遍固定种子、时间单位注释、事件实例重建检查这三步再做任何参数实验心里都有底。希望帮到你。本文还有配套的精品资源点击获取