simpy

SimPy - 离散事件仿真

概述

SimPy 是一个基于标准 Python 的进程式离散事件仿真框架。使用 SimPy 对系统建模，其中的实体（客户、车辆、数据包等）会随着时间相互交互，并竞争共享资源（服务器、机器、带宽等）。

核心功能：

• 使用 Python 生成器函数进行进程建模
• 共享资源管理（服务器、容器、存储区）
• 事件驱动的调度与同步
• 与挂钟时间同步的实时仿真
• 全面的监控与数据收集

何时使用此技能

在以下情况下使用 SimPy 技能：

1. 对离散事件系统建模 - 事件以不规则间隔发生的系统
2. 资源竞争 - 实体竞争有限资源（服务器、机器、员工）
3. 队列分析 - 研究等待队列、服务时间和吞吐量
4. 流程优化 - 分析制造、物流或服务流程
5. 网络仿真 - 数据包路由、带宽分配、延迟分析
6. 容量规划 - 确定满足预期性能所需的最佳资源水平
7. 系统验证 - 在实施前测试系统行为

不适用于：

• 使用固定时间步长的连续仿真（可考虑 SciPy ODE 求解器）
• 不共享资源的独立进程
• 纯数学优化（可考虑 SciPy optimize）

快速入门

基本仿真结构

import simpy

def process(env, name):
    """A simple process that waits and prints."""
    print(f'{name} starting at {env.now}')
    yield env.timeout(5)
    print(f'{name} finishing at {env.now}')

# Create environment
env = simpy.Environment()

# Start processes
env.process(process(env, 'Process 1'))
env.process(process(env, 'Process 2'))

# Run simulation
env.run(until=10)

资源使用模式

import simpy

def customer(env, name, resource):
    """Customer requests resource, uses it, then releases."""
    with resource.request() as req:
        yield req  # Wait for resource
        print(f'{name} got resource at {env.now}')
        yield env.timeout(3)  # Use resource
        print(f'{name} released resource at {env.now}')

env = simpy.Environment()
server = simpy.Resource(env, capacity=1)

env.process(customer(env, 'Customer 1', server))
env.process(customer(env, 'Customer 2', server))
env.run()

核心概念

1. 环境

仿真环境负责管理时间和调度事件。

import simpy

# Standard environment (runs as fast as possible)
env = simpy.Environment(initial_time=0)

# Real-time environment (synchronized with wall-clock)
import simpy.rt
env_rt = simpy.rt.RealtimeEnvironment(factor=1.0)

# Run simulation
env.run(until=100)  # Run until time 100
env.run()  # Run until no events remain

2. 进程

进程使用 Python 生成器函数（包含 yield 语句的函数）定义。

def my_process(env, param1, param2):
    """Process that yields events to pause execution."""
    print(f'Starting at {env.now}')

    # Wait for time to pass
    yield env.timeout(5)

    print(f'Resumed at {env.now}')

    # Wait for another event
    yield env.timeout(3)

    print(f'Done at {env.now}')
    return 'result'

# Start the process
env.process(my_process(env, 'value1', 'value2'))

3. 事件

事件是进程同步的基本机制。进程会让出事件，并在这些事件触发时恢复执行。

常见事件类型：

• env.timeout(delay) - 等待一段时间
• resource.request() - 请求资源
• env.event() - 创建自定义事件
• env.process(func()) - 将进程作为事件
• event1 & event2 - 等待所有事件（AllOf）
• event1 | event2 - 等待任一事件（AnyOf）

资源

SimPy 为不同场景提供多种资源类型。有关详细信息，请参阅 references/resources.md。

资源类型摘要

资源类型	用例
Resource	容量有限的资源（服务器、机器）
PriorityResource	基于优先级的排队
PreemptiveResource	高优先级任务可中断低优先级任务
Container	批量物料（燃料、水）
Store	Python 对象存储（FIFO）
FilterStore	按条件检索物品
PriorityStore	按优先级排序的物品

快速参考

import simpy

env = simpy.Environment()

# Basic resource (e.g., servers)
resource = simpy.Resource(env, capacity=2)

# Priority resource
priority_resource = simpy.PriorityResource(env, capacity=1)

# Container (e.g., fuel tank)
fuel_tank = simpy.Container(env, capacity=100, init=50)

# Store (e.g., warehouse)
warehouse = simpy.Store(env, capacity=10)

常见仿真模式

模式 1：客户-服务器队列

import simpy
import random

def customer(env, name, server):
    arrival = env.now
    with server.request() as req:
        yield req
        wait = env.now - arrival
        print(f'{name} waited {wait:.2f}, served at {env.now}')
        yield env.timeout(random.uniform(2, 4))

def customer_generator(env, server):
    i = 0
    while True:
        yield env.timeout(random.uniform(1, 3))
        i += 1
        env.process(customer(env, f'Customer {i}', server))

env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
env.process(customer_generator(env, server))
env.run(until=20)

模式 2：生产者-消费者

import simpy

def producer(env, store):
    item_id = 0
    while True:
        yield env.timeout(2)
        item = f'Item {item_id}'
        yield store.put(item)
        print(f'Produced {item} at {env.now}')
        item_id += 1

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

env = simpy.Environment()
store = simpy.Store(env, capacity=10)
env.process(producer(env, store))
env.process(consumer(env, store))
env.run(until=20)

模式 3：并行任务执行

import simpy

def task(env, name, duration):
    print(f'{name} starting at {env.now}')
    yield env.timeout(duration)
    print(f'{name} done at {env.now}')
    return f'{name} result'

def coordinator(env):
    # Start tasks in parallel
    task1 = env.process(task(env, 'Task 1', 5))
    task2 = env.process(task(env, 'Task 2', 3))
    task3 = env.process(task(env, 'Task 3', 4))

    # Wait for all to complete
    results = yield task1 & task2 & task3
    print(f'All done at {env.now}')

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

工作流指南

步骤 1：定义系统

确定：

• 实体：什么会在系统中流动？（客户、零件、数据包）
• 资源：有哪些约束？（服务器、机器、带宽）
• 进程：有哪些活动？（到达、服务、离开）
• 指标：要测量什么？（等待时间、利用率、吞吐量）

步骤 2：实现进程函数

为每种进程类型创建生成器函数：

def entity_process(env, name, resources, parameters):
    # Arrival logic
    arrival_time = env.now

    # Request resources
    with resource.request() as req:
        yield req

        # Service logic
        service_time = calculate_service_time(parameters)
        yield env.timeout(service_time)

    # Departure logic
    collect_statistics(env.now - arrival_time)

步骤 3：设置监控

使用监控工具收集数据。有关全面的技术说明，请参阅 references/monitoring.md。

from scripts.resource_monitor import ResourceMonitor

# Create and monitor resource
resource = simpy.Resource(env, capacity=2)
monitor = ResourceMonitor(env, resource, "Server")

# After simulation
monitor.report()

步骤 4：运行并分析

# Run simulation
env.run(until=simulation_time)

# Generate reports
monitor.report()
stats.report()

# Export data for further analysis
monitor.export_csv('results.csv')

高级功能

进程交互

进程可以通过事件、进程让出和中断进行交互。详细模式请参阅 references/process-interaction.md。

关键机制：

• 事件信号：用于协作的共享事件
• 进程让出：等待其他进程完成
• 中断：强制恢复进程以进行抢占

实时仿真

将仿真与挂钟时间同步，用于硬件在环或交互式应用。请参阅 references/real-time.md。

import simpy.rt

env = simpy.rt.RealtimeEnvironment(factor=1.0)  # 1:1 time mapping
# factor=0.5 means 1 sim unit = 0.5 seconds (2x faster)

全面监控

监控进程、资源和事件。有关相关技术，请参阅 references/monitoring.md，包括：

• 状态变量跟踪
• 资源猴子补丁
• 事件跟踪
• 统计数据收集

脚本和模板

basic_simulation_template.py

用于构建排队仿真的完整模板，包含：

• 可配置参数
• 统计数据收集
• 客户生成
• 资源使用
• 报告生成

用法：

from scripts.basic_simulation_template import SimulationConfig, run_simulation

config = SimulationConfig()
config.num_resources = 2
config.sim_time = 100
stats = run_simulation(config)
stats.report()

resource_monitor.py

可重复使用的监控工具：

• ResourceMonitor - 跟踪单个资源
• MultiResourceMonitor - 监控多个资源
• ContainerMonitor - 跟踪容器水位
• 自动计算统计数据
• CSV 导出功能

用法：

from scripts.resource_monitor import ResourceMonitor

monitor = ResourceMonitor(env, resource, "My Resource")
# ... run simulation ...
monitor.report()
monitor.export_csv('data.csv')

参考文档

特定主题的详细指南：

• references/resources.md - 所有资源类型及示例
• references/events.md - 事件系统和模式
• references/process-interaction.md - 进程同步
• references/monitoring.md - 数据收集技术
• references/real-time.md - 实时仿真设置

最佳实践

1. 生成器函数：始终在进程函数中使用 yield
2. 资源上下文管理器：使用 with resource.request() as req: 自动清理
3. 可复现性：设置 random.seed() 以获得一致的结果
4. 监控：在整个仿真过程中收集数据，而不只是在结束时收集
5. 验证：将简单案例与解析解进行比较
6. 文档：为进程逻辑和参数选择添加注释
7. 模块化设计：将进程逻辑、统计数据和配置分开

常见陷阱

1. 忘记使用 yield：进程必须让出事件才能暂停
2. 重复使用事件：事件只能触发一次
3. 资源泄漏：使用上下文管理器或确保释放资源
4. 阻塞操作：避免在进程中调用 Python 阻塞操作
5. 时间单位：保持时间单位解释的一致性
6. 死锁：确保至少有一个进程能够继续推进

示例用例

• 制造业：机器调度、生产线、库存管理
• 医疗保健：急诊室仿真、患者流转、人员分配
• 电信：网络流量、数据包路由、带宽分配
• 交通运输：交通流量、物流、车辆路线规划
• 服务运营：呼叫中心、零售结账、预约安排
• 计算机系统：CPU 调度、内存管理、I/O 操作