在动手写仿真代码之前，有三个关键问题必须先想清楚。否则后期调试会陷入“数据找不到、监控缺失、性能爆炸”的死循环：

• 监控什么——进程状态、资源利用率还是全局事件流？
• 什么时候监控——固定时间间隔还是状态变化时触发？
• 数据往哪放——内存、文件、数据库还是 NumPy/Pandas？

这三个答案几乎决定了整个监控方案的走向。SimPy 本身不提供“一键式”自动监控，它更像一个工具箱。把钩子和接口留给你，让你用 Python 原生的方式自由组合。

一、监控你自己的进程

最简单的列表收集

定义一个列表，状态变量有变化就 append 进去——就这么直接。

import simpy
data =
def test_process:
val = 0
for i in range:
val += env.now
data.append # 在此收集数据
yield env.timeout
env = simpy.Environment
p = env.process)
env.run
print
# 输出: Collected:

把 data 当作参数传入进程。而不是使用全局变量，这样可以避免很多潜在冲突。说起来，env.now 随时可读。每次 yield 前后都是插入采集逻辑的好时机。

记录多个字段

当需要记录多个字段时namedtuple/ 能让结构更清晰。

from collections import namedtuple
import simpy
Record = namedtuple
data =
def my_process:
state = 'idle'
for i in range:
state = 'busy'
data.append)
yield env.timeout
state = 'idle'
data.append)
yield env.timeout
env = simpy.Environment
env.process)
env.run
print
# 示例输出:

性能这件事，别等出问题再想

If your target storage is a database or a NumPy array,never write to it on every simulation step. Cache in a Python list first and convert in bulk after run.

import numpy as np
buffer = # 仿真期间只操作 Python list。速度极快
buffer.append) # O append
# 仿真结束后一次性转换为 NumPy 数组
arr = np.array(buffer,dtype=)
print

The difference 娱乐ween O list appends and O row‑wise inserts into a NumPy array becomes dramatic once you reach millions of events.

二、资源监控：Monkey Patching 登场

什么是 Monkey Patching

A runtime technique that replaces an object's method with a wrapper while preserving original functionality.

from functools import wraps
def patch_resource:
"""为 Resource 的关键方法注入前置/后置回调。"""
def wrapper_factory:
@wraps
def wrapper:
if pre:
pre
ret = func
if post:
post
return ret
return wrapper
for name in :
if hasattr:
setattr(resource,name,wrapper_factory))

@wraps 的作用 

The decorator copies metadata 
preventing every wrapped method from being displayed as “wrapper”. This makes debugging far easier.

# 没有 @wraps:
wrapper.__name__ == 'wrapper'
# 有 @wraps:
wrapper.__name__ == 'request' # 与原函数名一致

完整的资源监控示例 

from functools import partial
import simpy
def patch_resource:
def get_wrapper:
@wraps
def wrapper:
if pre:
pre
ret = func
if post:
post
return ret
return wrapper
for name in :
if hasattr:
setattr(resource,name,get_wrapper))
def monitor:
item = (resource._env.now,resource.count,len)
data.append
def user:
with resource.request as req:
yield req
print
yield env.timeout
env = simpy.Environment
resource = simpy.Resource
data =
patch_resource(resource,post=partial(monitor,data))
for i in range:
env.process(user(env,f'User-{i}',resource))
env.run
至于print。')
for row in data:
print

各类资源的可监控属性 



资源类型
关键属性
含义



.Resource
.count
当前占用数量 

.capacityTotal capacity of resource 

.queueA list of pending requests 

.Container.level

  .Store 
 .items
 当前存储的物品列表

  .PreemptiveResource
 .users
 正在使用且带优先级的实体集合





三、全局事件追踪：在 Environment 层面动手脚 

子类化 Environment 

You can inherit from  simpy.Environment  and override its  step  method. Each time an event is processed,step is called – perfect for logging entire event stream.

import simpy
class MonitoredEnvironment:
def __init__:
super.__init__
self.event_log =
def step:
if self._queue:
event_time,_,_,event = self._queue
self.event_log.append({
'time' : event_time,'event_type': type.__name__
})
super.step
env = MonitoredEnvironment
def simple_proc:
yield env.timeout
yield env.timeout
env.process)
env.run
print


 functools.partial  的妙用 

You often need to bind extra arguments to a monitoring callback.  partial  creates a new callable with those arguments already set.

# 不用 partial，用 lambda
post=lambda r: monitor
# 用 partial，更简洁
post=partial


The two forms are equivalent;however when you combine several callbacks readability of  partial  shines.

四、定时采样还是事件驱动？

No universal answer – pick what fits your analysis needs.

 定时采样 

Simplifies downstream time‑series analysis because timestamps are uniform.

def monitor_process:
while True:
data.append({
'time' : env.now,'utilization': res.count/res.capacity,'queue_len' : len。})
yield env.timeout
env = simpy.Environment
res = simpy.Resource
data =
env.process)
env.run


The downside: short spikes that occur 娱乐ween two samples will be missed.

 事件驱动采样 

This is exactly what Monkey Patching provides – record on every state change. You get complete information but timestamps are irregular;you may need extra post‑processing to compute time‑weighted averages.

A common hybrid approach:


Use event‑driven hooks to capture precise transitions. 

Run a low‑frequency sampling process to generate smooth series for plotting. 


五、数据往哪放 

内存 

data=
data.append)


文件 

import csv
with open as f:
writer=csv.writer
writer.writerow
writer.writerows # 仿真结束后一次性写入


NumPy / Pandas 

import pandas as pd
df=pd.DataFrame(data,columns=)
print)
df.plot
avg_util=df.mean/capacity


数据库 

import sqlite3
conn=sqlite3.connect
conn.executemany',data) # 批量插入，避免频繁 I/O
conn.commit


六、综合实战：一个完整的服务台监控程序 

from functools import partial,wraps
from collections import namedtuple
import simpy,random,pandas as pd
Record=namedtuple
resource_log=
process_log=
def patch_resource:
def get_wrapper:
@wraps
def wrapper:
if pre: pre
r=func
if post: post
return r
return wrapper
for n in :
if hasattr:
setattr))
def resource_monitor:
data.append(Record(res._env.now,res.count。len))
def customer:
arrive=env.now
with srv.request as req:
yield req # 等待资源分配
wait=env.now-arrive # 等待时间
service=random.expovariate# 服务时间随机分布
yield env.timeout
p_log.append
def arrivals:
cid=0
while True:
yield env.timeout)
env.process)
cid+=1
# 环境与资源初始化
env=simpy.Environment
server=simpy.Resource
# 打上监控补丁
patch_resource)
# 启动到达进程并运行仿真
env.process)
env.run
# 后处理：计算平均等待时间 & 平均利用率
avg_wait=sum/len
avg_util=sum/*server.capacity)
print
print
print} Customers:{len}')


从收尾来看，四个工具，打遍天下 




 技术 
 场景 
 主要机制 <
/ tr>


/list append/
/进程内部直接采集/
/O 插入 /

/Monkey Patching/
/资源使用细粒度监控/
/包装 request/release + @wraps /

/定时采样进程/
/均匀时间序列 /
/while True + env.timeout /

/子类化 Environment/
/全局事件追踪/
/重写 step /
/ tbody>/ table /







再看养成习惯，永远先把数据缓存在 Python list 中，仿真结束后再批量转换为 CSV、NumPy 或数据库表。

Citations：


Simp​y 官方文档 — Monitoring：

Simp​y 官方文档 — Overview：

The Winterflower Blog — What is going on in this SimPy Monitoring example?: