simpy-discrete-event-simulation

SimPy — 离散事件仿真

概览

SimPy 是一个基于进程的离散事件仿真框架，使用标准 Python 生成器。它可以模拟实体（客户、车辆、数据包）随时间与共享资源（服务器、机器、带宽）交互的系统，并提供事件驱动调度和可选的实时同步。

适用场景

• 模拟存在资源竞争（服务器、机器、员工）的排队系统
• 制造流程仿真（生产线、调度、瓶颈分析）
• 网络仿真（数据包路由、带宽分配、延迟分析）
• 容量规划（确定满足目标吞吐量的最佳资源规模）
• 医疗运营（急诊患者流转、员工分配、床位管理）
• 物流与运输（仓库运营、车辆路径规划）
• 对于连续时间 ODE 系统 → 使用 SciPy solve_ivp
• 对于基于 Agent 的建模 → 使用 Mesa

前置条件

# pip install simpy
import simpy
import random

快速开始

import simpy
import random

def customer(env, name, server):
    """Customer arrives, waits for server, gets served, departs."""
    arrival = env.now
    with server.request() as req:
        yield req  # Wait in queue
        wait = env.now - arrival
        yield env.timeout(random.expovariate(1/3))  # Service time
        print(f'{name}: waited {wait:.1f}, served at {env.now:.1f}')

def arrivals(env, server):
    for i in range(20):
        yield env.timeout(random.expovariate(1/2))  # Inter-arrival
        env.process(customer(env, f'C{i}', server))

env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
env.process(arrivals(env, server))
env.run(until=50)

核心 API

1. 环境与进程

import simpy

# Standard environment
env = simpy.Environment(initial_time=0)

# Processes are Python generators that yield events
def machine(env, name, repair_time):
    while True:
        yield env.timeout(random.expovariate(1/10))  # Time to failure
        print(f'{name} broke at {env.now:.1f}')
        yield env.timeout(repair_time)
        print(f'{name} repaired at {env.now:.1f}')

# Start processes — returns a Process event
proc = env.process(machine(env, 'Machine-1', repair_time=2))

# Run until time limit or no events remain
env.run(until=100)
# env.run()  # Run until no more events

# Current simulation time
print(f'Final time: {env.now}')

# Processes can return values and be awaited
def subtask(env, duration):
    yield env.timeout(duration)
    return f'completed in {duration}'

def main_task(env):
    # Sequential: wait for one process
    result = yield env.process(subtask(env, 5))
    print(f'Subtask {result} at {env.now}')

    # Parallel: wait for ALL (AllOf)
    t1 = env.process(subtask(env, 3))
    t2 = env.process(subtask(env, 4))
    results = yield t1 & t2  # AllOf — resumes when both done
    print(f'Both done at {env.now}')

    # Race: wait for ANY (AnyOf)
    t3 = env.process(subtask(env, 2))
    t4 = env.process(subtask(env, 6))
    result = yield t3 | t4  # AnyOf — resumes when first completes
    print(f'First done at {env.now}')

env = simpy.Environment()
env.process(main_task(env))
env.run()

2. 资源

import simpy

env = simpy.Environment()

# Basic resource — capacity-limited (e.g., 2 servers)
server = simpy.Resource(env, capacity=2)
print(f'Capacity: {server.capacity}, In use: {server.count}, Queue: {len(server.queue)}')

# Priority resource — lower number = higher priority
priority_server = simpy.PriorityResource(env, capacity=1)

def vip_customer(env, res):
    with res.request(priority=1) as req:  # Higher priority
        yield req
        yield env.timeout(3)

def regular_customer(env, res):
    with res.request(priority=10) as req:  # Lower priority
        yield req
        yield env.timeout(3)

# Preemptive resource — high priority interrupts low priority
preemptive = simpy.PreemptiveResource(env, capacity=1)

def urgent_job(env, res):
    with res.request(priority=0, preempt=True) as req:
        yield req  # May interrupt current user
        yield env.timeout(1)

# Container — bulk material (fuel, water, inventory)
tank = simpy.Container(env, capacity=100, init=50)

def refuel(env, tank):
    yield tank.put(30)  # Add 30 units
    print(f'Tank level: {tank.level}/{tank.capacity}')

def consume(env, tank):
    yield tank.get(20)  # Remove 20 units
    print(f'Tank level: {tank.level}/{tank.capacity}')

# Store — FIFO object storage
warehouse = simpy.Store(env, capacity=10)

def producer(env, store):
    for i in range(5):
        yield env.timeout(2)
        yield store.put(f'Item-{i}')

def consumer(env, store):
    while True:
        item = yield store.get()
        print(f'Got {item} at {env.now}')
        yield env.timeout(3)

# FilterStore — selective retrieval
parts = simpy.FilterStore(env, capacity=20)

def picker(env, store):
    # Get specific item matching condition
    item = yield store.get(lambda x: x['color'] == 'red')
    print(f'Found red item: {item}')

3. 事件与同步

import simpy

env = simpy.Environment()

# Basic event — manual trigger for signaling between processes
signal = env.event()

def waiter(env, event):
    print(f'Waiting at {env.now}')
    value = yield event  # Blocks until triggered
    print(f'Got signal "{value}" at {env.now}')

def sender(env, event):
    yield env.timeout(5)
    event.succeed(value='go')  # Trigger with value

env.process(waiter(env, signal))
env.process(sender(env, signal))
env.run()
# Output: Waiting at 0, Got signal "go" at 5

# Timeout — most common event
yield env.timeout(delay=5)

# Process interruption
def interruptible(env, name):
    try:
        yield env.timeout(10)
    except simpy.Interrupt as interrupt:
        print(f'{name} interrupted: {interrupt.cause} at {env.now}')

def interruptor(env, proc):
    yield env.timeout(3)
    proc.interrupt('maintenance')

proc = env.process(interruptible(env, 'Worker'))
env.process(interruptor(env, proc))

# Barrier synchronization — wait for N processes
class Barrier:
    def __init__(self, env, n):
        self.env = env
        self.n = n
        self.count = 0
        self.event = env.event()

    def wait(self):
        self.count += 1
        if self.count >= self.n:
            self.event.succeed()
        return self.event

def phase_worker(env, name, barrier):
    yield env.timeout(random.uniform(1, 5))  # Phase work
    print(f'{name} reached barrier at {env.now:.1f}')
    yield barrier.wait()  # Wait for all workers
    print(f'{name} passed barrier at {env.now:.1f}')

env = simpy.Environment()
barrier = Barrier(env, n=3)
for i in range(3):
    env.process(phase_worker(env, f'W{i}', barrier))
env.run()

4. 监控与统计

import simpy

# Inline statistics collection
class Stats:
    def __init__(self):
        self.wait_times = []
        self.queue_lengths = []

    def report(self):
        if self.wait_times:
            avg_wait = sum(self.wait_times) / len(self.wait_times)
            max_wait = max(self.wait_times)
            print(f'Avg wait: {avg_wait:.2f}, Max wait: {max_wait:.2f}')
            print(f'Customers served: {len(self.wait_times)}')

def customer(env, name, server, stats):
    arrival = env.now
    with server.request() as req:
        yield req
        wait = env.now - arrival
        stats.wait_times.append(wait)
        stats.queue_lengths.append(len(server.queue))
        yield env.timeout(random.expovariate(1/3))

env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
stats = Stats()

def gen(env, server, stats):
    for i in range(100):
        yield env.timeout(random.expovariate(1/2))
        env.process(customer(env, f'C{i}', server, stats))

env.process(gen(env, server, stats))
env.run(until=200)
stats.report()

# Resource monitoring via monkey-patching
def patch_resource(resource, data):
    """Patch resource to log request/release events."""
    original_request = resource.request
    original_release = resource.release

    def monitored_request(*args, **kwargs):
        req = original_request(*args, **kwargs)
        data.append((resource._env.now, 'request', resource.count, len(resource.queue)))
        return req

    def monitored_release(*args, **kwargs):
        result = original_release(*args, **kwargs)
        data.append((resource._env.now, 'release', resource.count, len(resource.queue)))
        return result

    resource.request = monitored_request
    resource.release = monitored_release

log = []
patch_resource(server, log)
# After simulation: analyze log for utilization, queue dynamics

5. 实时仿真

import simpy.rt

# Real-time environment — synchronized with wall clock
env = simpy.rt.RealtimeEnvironment(factor=1.0)  # 1 sim unit = 1 second
# factor=0.1 → 10x faster (1 sim unit = 0.1 seconds)
# factor=60 → 1 sim unit = 1 minute

# Strict mode raises RuntimeError if simulation can't keep up
env_strict = simpy.rt.RealtimeEnvironment(factor=1.0, strict=True)

# Non-strict mode (default) allows slower-than-real-time execution
env_relaxed = simpy.rt.RealtimeEnvironment(factor=1.0, strict=False)

def periodic_task(env, interval):
    while True:
        print(f'Tick at sim time {env.now:.1f}')
        yield env.timeout(interval)

env = simpy.rt.RealtimeEnvironment(factor=1.0)
env.process(periodic_task(env, 2.0))
env.run(until=10)
# Prints "Tick" every ~2 real seconds

核心概念

资源选择指南

需求	资源类型	主要特性
有限的服务器／机器	Resource	FIFO 队列，容量限制
优先级排队	PriorityResource	数值越小，优先级越高
可抢占调度	PreemptiveResource	高优先级会中断当前使用者
散装材料（燃料、水）	Container	put(amount) / get(amount)，连续变化的存量
对象队列（FIFO）	Store	put(item) / get()，按序取出
条件检索	FilterStore	get(lambda x: condition)
按优先级排序的项	PriorityStore	按优先级对项排序

进程交互机制

机制	适用场景	代码模式
事件信号	向多个等待者广播	event = env.event() → yield event / event.succeed()
进程让出控制权	顺序或并行执行	yield env.process(func()) 或 yield p1 & p2
中断	抢占、维护、取消	proc.interrupt(cause) + try/except simpy.Interrupt
超时竞争	超时并取消	`yield event

常见工作流

1. 制造流水线仿真

import simpy
import random

def part(env, name, machines, buffer, stats):
    """Part flows through sequential machines with intermediate buffer."""
    for i, machine in enumerate(machines):
        with machine.request() as req:
            yield req
            process_time = random.triangular(1, 3, 2)
            yield env.timeout(process_time)

    if buffer.level < buffer.capacity:
        yield buffer.put(1)
        stats['produced'] += 1

def part_generator(env, machines, buffer, stats):
    i = 0
    while True:
        yield env.timeout(random.expovariate(1/2))
        env.process(part(env, f'Part-{i}', machines, buffer, stats))
        i += 1

random.seed(42)
env = simpy.Environment()
machines = [simpy.Resource(env, capacity=1) for _ in range(3)]
output_buffer = simpy.Container(env, capacity=100, init=0)
stats = {'produced': 0}
env.process(part_generator(env, machines, output_buffer, stats))
env.run(until=480)  # 8-hour shift
print(f'Parts produced: {stats["produced"]}')
print(f'Buffer level: {output_buffer.level}')

2. 带优先级的多服务器队列

import simpy
import random

def patient(env, name, priority, er, stats):
    arrival = env.now
    with er.request(priority=priority) as req:
        yield req
        wait = env.now - arrival
        stats['waits'].append((name, priority, wait))
        service = random.expovariate(1/15)  # ~15 min avg
        yield env.timeout(service)

def patient_arrivals(env, er, stats):
    i = 0
    while True:
        yield env.timeout(random.expovariate(1/5))  # ~5 min between arrivals
        pri = random.choices([1, 2, 3], weights=[0.1, 0.3, 0.6])[0]
        env.process(patient(env, f'P{i}', pri, er, stats))
        i += 1

random.seed(42)
env = simpy.Environment()
er = simpy.PriorityResource(env, capacity=3)
stats = {'waits': []}
env.process(patient_arrivals(env, er, stats))
env.run(until=480)

# Analyze by priority
for pri in [1, 2, 3]:
    waits = [w for _, p, w in stats['waits'] if p == pri]
    if waits:
        print(f'Priority {pri}: avg wait {sum(waits)/len(waits):.1f}, n={len(waits)}')

3. 带监控的生产者－消费者

纯文本工作流（结合核心 API 模块 2、3、4）：

1. 创建具有有限容量的 simpy.Store（模块 2：资源）
2. 实现生产者进程，通过 yield store.put(item) 并设置生产延迟（模块 2）
3. 实现消费者进程，通过 yield store.get() 并设置处理延迟（模块 2）
4. 添加事件信号，在存储已满时实施背压（模块 3：事件）
5. 收集吞吐量、队列长度和空闲时间统计数据（模块 4：监控）
6. 运行仿真并生成报告

关键参数

参数	模块	默认值	范围	效果
capacity	Resource	1	1–∞	并发使用者数量
priority	PriorityResource.request	0	int	数值越小，优先级越高
preempt	PreemptiveResource.request	True	bool	是否中断低优先级使用者
capacity	Container	float('inf')	0–∞	最大存量
init	Container	0	0–capacity	初始存量
capacity	Store	float('inf')	0–∞	最大项数
factor	RealtimeEnvironment	1.0	>0	仿真时间与墙上时间之比
strict	RealtimeEnvironment	False	bool	落后于计划时是否抛出错误
initial_time	Environment	0	any float	仿真开始时间

最佳实践

1. 始终使用上下文管理器管理资源：with resource.request() as req: yield req 可确保即使发生异常也会自动释放资源
2. 设置随机种子以确保可复现性：创建进程前调用 random.seed(42)；如需更多分布类型，请使用 numpy.random
3. 在模拟过程中收集统计数据：模拟期间将数据追加到列表中，并在 env.run() 之后计算汇总值——不要在模拟过程中查询
4. 为流程时间使用三角分布：对于服务时间，random.triangular(min, max, mode) 比均匀分布更贴近实际
5. 反模式——忘记使用 yield：不带 yield 调用 env.timeout(5) 只会创建事件，不会暂停进程。始终使用 yield env.timeout(5)
6. 反模式——重复使用事件：事件只能触发一次。每个信号周期都要创建新的 env.event()；对于可重复的信号，请在循环中创建新事件

常用示例

示例：多次重复模拟

import simpy
import random
import statistics

def run_single(seed, sim_time=480, n_servers=2):
    random.seed(seed)
    env = simpy.Environment()
    server = simpy.Resource(env, capacity=n_servers)
    waits = []

    def customer(env, server):
        arrival = env.now
        with server.request() as req:
            yield req
            waits.append(env.now - arrival)
            yield env.timeout(random.expovariate(1/3))

    def gen(env, server):
        while True:
            yield env.timeout(random.expovariate(1/2))
            env.process(customer(env, server))

    env.process(gen(env, server))
    env.run(until=sim_time)
    return sum(waits) / len(waits) if waits else 0

# Run 30 replications
results = [run_single(seed=i) for i in range(30)]
print(f'Mean avg wait: {statistics.mean(results):.2f}')
print(f'95% CI: ±{1.96 * statistics.stdev(results) / len(results)**0.5:.2f}')

示例：基于中断的维护

import simpy
import random

def machine(env, name, repair_crew):
    while True:
        try:
            # Operate until failure
            ttf = random.expovariate(1/50)  # Mean 50 time units to failure
            yield env.timeout(ttf)
            print(f'{name} failed at {env.now:.1f}')
        except simpy.Interrupt:
            print(f'{name} interrupted for maintenance at {env.now:.1f}')

        # Repair (needs repair crew)
        with repair_crew.request() as req:
            yield req
            repair = random.uniform(2, 5)
            yield env.timeout(repair)
            print(f'{name} repaired at {env.now:.1f}')

def maintenance_scheduler(env, machines_procs):
    """Periodic preventive maintenance every 40 time units."""
    while True:
        yield env.timeout(40)
        for proc in machines_procs:
            if proc.is_alive:
                proc.interrupt('scheduled maintenance')

env = simpy.Environment()
repair_crew = simpy.Resource(env, capacity=1)
procs = [env.process(machine(env, f'M{i}', repair_crew)) for i in range(3)]
env.process(maintenance_scheduler(env, procs))
env.run(until=200)

示例：基于容器的供应链

import simpy
import random

def supplier(env, warehouse):
    """Deliver batch when level drops below reorder point."""
    while True:
        if warehouse.level < 20:  # Reorder point
            yield env.timeout(random.uniform(5, 10))  # Lead time
            amount = min(50, warehouse.capacity - warehouse.level)
            yield warehouse.put(amount)
            print(f'Delivered {amount} units at {env.now:.1f}, level={warehouse.level}')
        yield env.timeout(1)  # Check interval

def demand(env, warehouse, stats):
    while True:
        yield env.timeout(random.expovariate(1/2))
        qty = random.randint(1, 5)
        if warehouse.level >= qty:
            yield warehouse.get(qty)
            stats['fulfilled'] += qty
        else:
            stats['stockouts'] += 1

env = simpy.Environment()
warehouse = simpy.Container(env, capacity=100, init=80)
stats = {'fulfilled': 0, 'stockouts': 0}
env.process(supplier(env, warehouse))
env.process(demand(env, warehouse, stats))
env.run(until=500)
print(f'Fulfilled: {stats["fulfilled"]}, Stockouts: {stats["stockouts"]}')

故障排除

问题	原因	解决方案
进程未暂停	事件前缺少 yield	始终使用 yield env.timeout(x)，不要只写 env.timeout(x)
RuntimeError: Event already triggered	重复使用已触发的事件	每个信号周期都创建新的 env.event()
资源始终未释放	未使用上下文管理器	使用 with resource.request() as req: 模式
模拟无限运行	未设置 until 参数，且进程无限运行	添加 env.run(until=time) 或确保进程终止
未捕获 simpy.Interrupt	可中断进程中缺少 try/except	使用 try: ... except simpy.Interrupt: 包裹 yield
队列顺序错误	使用了 Resource 而非 PriorityResource	优先级排队时改用 simpy.PriorityResource
实时模拟太慢	计算超出实际时间预算	设置 strict=False 或增大 factor
容器的 put 操作阻塞	容器已满	在执行 put 前检查 container.level < container.capacity
FilterStore 的 get 操作一直阻塞	没有匹配的项	确保生产者创建的项符合筛选条件
统计数据为空	在 env.run() 之前收集数据	env.run() 完成后调用 stats.report()

附带资源

• references/process_events_guide.md — 详细介绍事件生命周期（触发→处理）、复合事件（AllOf/AnyOf）、进程交互模式（信号、屏障、中断、握手）及高级同步。整合自原始的 events.md（375 行）和 process-interaction.md（425 行）
• references/resources_monitoring_guide.md — 完整介绍资源类型（Resource、Priority、Preemptive、Container、Store、FilterStore、PriorityStore）、通过 monkey-patching 进行监控（ResourceMonitor、ContainerMonitor 类）、统计数据收集模式、CSV/matplotlib 导出及实时模拟（RealtimeEnvironment、时间缩放、严格模式、HIL 模式）。整合自原始的 resources.md（276 行）、monitoring.md（476 行）和 real-time.md（396 行）。脚本功能（basic_simulation_template.py、resource_monitor.py）已整合到 Core API 监控示例和常用示例中

相关技能

• matplotlib-scientific-plotting — 可视化模拟结果（队列长度、随时间变化的利用率）
• polars-dataframes — 分析大型模拟输出数据集

参考资料

• SimPy 文档：https://simpy.readthedocs.io/
• SimPy GitHub：https://github.com/teamhide/simpy
• Banks 等人，《离散事件系统模拟》（教材参考）