Python中基于过程的离散事件仿真框架。当构建包含过程、队列、资源和时间事件（如制造系统、服务运营、网络流量、物流或任何实体随时间与共享资源交互的系统）的仿真时使用此技能。

SimPy - 离散事件模拟

概述

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

核心功能：

使用 Python 生成器函数进行过程建模
共享资源管理（服务器、容器、存储）
事件驱动的调度和同步
与 wall-clock 时间同步的实时模拟
全面的监控和数据收集

何时使用此技能

当以下情况时使用 SimPy 技能：

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

不适合：

固定时间步长的连续模拟（考虑 SciPy ODE 求解器）
无资源共享的独立过程
纯数学优化（考虑 SciPy optimize）

快速入门

基本模拟结构

import simpy

def process(env, name):
    """一个简单的等待和打印过程。"""
    print(f'{name} starting at {env.now}')
    yield env.timeout(5)
    print(f'{name} finishing at {env.now}')

# 创建环境
env = simpy.Environment()

# 启动过程
env.process(process(env, 'Process 1'))
env.process(process(env, 'Process 2'))

# 运行模拟
env.run(until=10)

资源使用模式

import simpy

def customer(env, name, resource):
    """客户请求资源，使用它，然后释放。"""
    with resource.request() as req:
        yield req  # 等待资源
        print(f'{name} got resource at {env.now}')
        yield env.timeout(3)  # 使用资源
        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

# 标准环境（尽可能快地运行）
env = simpy.Environment(initial_time=0)

# 实时环境（与 wall-clock 同步）
import simpy.rt
env_rt = simpy.rt.RealtimeEnvironment(factor=1.0)

# 运行模拟
env.run(until=100)  # 运行到时间 100
env.run()  # 运行直到没有事件剩余

2. 过程

过程使用 Python 生成器函数（带有 yield 语句的函数）定义。

def my_process(env, param1, param2):
    """产生事件以暂停执行的过程。"""
    print(f'Starting at {env.now}')

    # 等待时间流逝
    yield env.timeout(5)

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

    # 等待另一个事件
    yield env.timeout(3)

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

# 启动过程
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()

# 基本资源（例如，服务器）
resource = simpy.Resource(env, capacity=2)

# 优先级资源
priority_resource = simpy.PriorityResource(env, capacity=1)

# 容器（例如，油箱）
fuel_tank = simpy.Container(env, capacity=100, init=50)

# 存储（例如，仓库）
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):
    # 并行启动任务
    task1 = env.process(task(env, 'Task 1', 5))
    task2 = env.process(task(env, 'Task 2', 3))
    task3 = env.process(task(env, 'Task 3', 4))

    # 等待所有完成
    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_time = env.now

    # 请求资源
    with resource.request() as req:
        yield req

        # 服务逻辑
        service_time = calculate_service_time(parameters)
        yield env.timeout(service_time)

    # 离开逻辑
    collect_statistics(env.now - arrival_time)

步骤 3：设置监控

使用监控工具收集数据。有关综合技术，请参阅 references/monitoring.md。

from scripts.resource_monitor import ResourceMonitor

# 创建并监控资源
resource = simpy.Resource(env, capacity=2)
monitor = ResourceMonitor(env, resource, "Server")

# 模拟后
monitor.report()

步骤 4：运行和分析

# 运行模拟
env.run(until=simulation_time)

# 生成报告
monitor.report()
stats.report()

# 导出数据以进行进一步分析
monitor.export_csv('results.csv')

高级功能

过程交互

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

关键机制：

事件信号：用于协调的共享事件
过程产生：等待其他过程完成
中断：强制恢复过程以进行抢占

实时模拟

将模拟与 wall-clock 时间同步，用于硬件在环或交互式应用程序。请参阅 references/real-time.md。

import simpy.rt

env = simpy.rt.RealtimeEnvironment(factor=1.0)  # 1:1 时间映射
# factor=0.5 表示 1 模拟单位 = 0.5 秒（快 2 倍）

综合监控

监控过程、资源和事件。有关技术，请参阅 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")
# ... 运行模拟 ...
monitor.report()
monitor.export_csv('data.csv')

参考文档

特定主题的详细指南：

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

最佳实践

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

常见陷阱

忘记 yield：过程必须产生事件以暂停
事件重用：事件只能触发一次
资源泄漏：使用上下文管理器或确保释放
阻塞操作：避免在过程中使用 Python 阻塞调用
时间单位：保持时间单位解释的一致性
死锁：确保至少一个过程可以取得进展

示例用例

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

SimPy - 离散事件模拟