通过机器车间案例解析SimPy核心概念

SimPy作为Python离散事件仿真框架，能高效模拟工厂、医院等复杂场景的并发流程。本文通过经典机器车间案例，详解其核心机制与实现方法。

先说清楚这是在做什么

SimPy 是一个用 Python 写离散事件仿真的框架。听起来学术，但本质很朴素——你想模拟"一堆事情同时发生、互相影响"的场景，比如工厂、医院、网络服务器，SimPy 就是干这个的。

它的核心设计很聪明：不用线程，不用回调地狱，而是借助 Python 原生的**生成器（generator）**来模拟并发。每个 yield 就是一次暂停，把控制权还给仿真调度器，等时机到了再继续。这个设计让代码读起来像在描述真实流程，而不是在跟框架搏斗。

官方文档里有一个"机器车间"的例子，场景不复杂，但把 SimPy 最核心的几个机制全都用上了：进程、超时、中断、抢占资源。把这一个例子吃透，SimPy 基本上就入门了。

---

场景是这样的

车间里有 10 台机器，不停地加工零件。每台机器会随机发生故障，故障了就得等维修工来修。问题是，维修工只有一个，他平时还有些杂活要干。机器一坏，杂活立刻靠边站，维修工优先去修机器。仿真跑满 4 周，最后统计每台机器生产了多少零件。

这个场景里有几个值得玩味的地方：机器的生产和故障是两个并发进程；维修工是一个有优先级的共享资源；故障发生时需要打断正在进行的生产。这三件事，恰好对应 SimPy 的三个核心机制。

---

核心概念，先建立一张地图

在看代码之前，把几个概念捋清楚，后面读起来会顺很多。

概念	对应类/方法	一句话说清楚
仿真环境	simpy.Environment()	仿真的时钟和调度器，所有事件都在里面跑
进程	env.process(generator)	把一个生成器函数注册成仿真进程
超时事件	env.timeout(delay)	让进程"睡"一段仿真时间
普通资源	simpy.Resource	有容量限制，先来先得
抢占资源	simpy.PreemptiveResource	高优先级可以插队，甚至踢走低优先级
中断	process.interrupt()	强行叫醒一个正在睡觉的进程
请求资源	resource.request()	申请占用资源，配合 with 自动释放

有一个心智模型贯穿始终：SimPy 里每个 yield 都是一个暂停点。进程在此挂起，把控制权交回给仿真环境，等事件触发后再接着跑。理解了这一点，后面所有的逻辑都会变得自然。

---

代码逐行拆解

参数定义：把现实世界翻译成数字

import random   # 生成随机数，模拟加工时间和故障时间
import simpy    # SimPy 仿真框架本体RANDOM_SEED  = 42         # 随机种子，固定它才能复现同样的结果
PT_MEAN      = 10.0       # 每个零件的平均加工时间（分钟）
PT_SIGMA     = 2.0        # 加工时间的波动幅度（正态分布标准差）
MTTF         = 300.0      # 平均故障间隔时间（分钟），Mean Time To Failure
BREAK_MEAN   = 1 / MTTF   # 指数分布的参数 λ，故障率
REPAIR_TIME  = 30.0       # 每次维修耗时（分钟，固定值）
JOB_DURATION = 30.0       # 维修工杂务的单次持续时间（分钟）
NUM_MACHINES = 10         # 车间机器数量
WEEKS        = 4          # 仿真运行周数
SIM_TIME     = WEEKS * 7 * 24 * 60  # 换算成分钟：40320 分钟

这里有个细节：BREAK_MEAN = 1 / MTTF 是指数分布的参数，而不是故障间隔本身。指数分布天然无记忆性，很适合模拟随机故障——机器不会"越用越容易坏"，每一刻的故障概率都一样。

---

随机时间生成：两种分布，两种用途

def time_per_part():
    """返回加工一个零件的实际耗时"""
    t = random.normalvariate(PT_MEAN, PT_SIGMA)  # 正态分布采样
    while t <= 0:
        # 正态分布有极小概率采出负数，循环重采直到得到正值
        t = random.normalvariate(PT_MEAN, PT_SIGMA)
    return tdef time_to_failure():
    """返回机器下次故障前的运行时间"""
    # 指数分布本身只产生正数，不需要额外过滤
    return random.expovariate(BREAK_MEAN)

加工时间用正态分布——有个均值，有些波动，符合真实工序的直觉。故障间隔用指数分布——随机、无记忆，符合设备随机失效的统计规律。两种分布，各司其职。

---

Machine 类：一台机器，两个并发进程

这是整个例子的核心。一台机器同时跑两个进程：一个负责生产，一个负责触发故障。两者并行，互相影响。

class Machine:
    def __init__(self, env, name, repairman):
        self.env        = env        # 保存仿真环境引用
        self.name       = name       # 机器名称，如 "Machine 0"
        self.parts_made = 0          # 已生产零件数，初始为 0
        self.broken     = False      # 当前是否处于故障状态        # 启动"生产"进程，并保存句柄——后面 break_machine 要用它来发中断
        self.process = env.process(self.working(repairman))        # 启动"故障触发"进程，不需要保存句柄，它自己会循环
        env.process(self.break_machine())

env.process() 是 SimPy 的入口，把一个生成器函数变成仿真进程。两个进程在 __init__ 里同时启动，从仿真一开始就并行运行。

---

生产进程：在等待中完成工作，在中断中处理故障

def working(self, repairman):
    """持续生产零件的主进程"""
    while True:                          # 仿真结束前永不停歇        done_in = time_per_part()        # 随机生成当前零件的加工时长        while done_in:                   # done_in > 0 说明零件还没做完
            start = self.env.now         # 记录本段加工的起始时刻            try:
                #  核心：yield timeout 让进程"睡眠" done_in 分钟
                # 仿真时钟向前推进，其他进程趁机运行
                yield self.env.timeout(done_in)                done_in = 0              # 正常醒来：零件做完了，退出内层循环            except simpy.Interrupt:
                #  被中断：break_machine 调用了 self.process.interrupt()
                # 执行流从 yield 处跳到这里，零件没做完                self.broken = True       # 标记为故障状态                # 算出还剩多少时间没加工（已过去 env.now - start 分钟）
                done_in -= self.env.now - start                #  向维修工申请服务，优先级 1（数字越小越优先）
                # with 语句保证用完自动释放，不会忘记还资源
                with repairman.request(priority=1) as req:
                    yield req                           # 等维修工空闲
                    yield self.env.timeout(REPAIR_TIME) # 等维修完成                self.broken = False      # 修好了，重置状态
                # 内层 while 继续，用剩余的 done_in 时间把零件做完        self.parts_made += 1             # 零件完工，计数 +1

这段代码的结构很有意思：用 try/except 来区分"正常完成"和"被打断"两条路径，而不是用 if/else 轮询状态。这是 SimPy 的惯用写法，读起来逻辑很清晰——正常情况走 try，异常情况走 except，和 Python 本身的异常处理哲学一脉相承。

---

故障触发进程：安静地等，然后出手

def break_machine(self):
    """周期性触发机器故障"""
    while True:        # 等待随机的故障间隔时间
        yield self.env.timeout(time_to_failure())        if not self.broken:
            # 只在机器正常运行时才触发故障
            # 如果机器已经在修了，就跳过这次，等下一次
            self.process.interrupt()  #  向 working 进程发送中断信号

self.broken 这个标志很关键。如果不判断，可能出现机器正在维修、故障进程又发来一次中断的情况，逻辑就乱了。这个小细节，是防御性编程的体现。

---

维修工杂务：被打断，然后继续

def other_jobs(env, repairman):
    """维修工的低优先级杂务"""
    while True:        done_in = JOB_DURATION           # 当前杂务的剩余时间        while done_in:                   # 没做完就一直重试
            #  优先级 2，低于机器维修的优先级 1
            # 如果维修工正在做杂务，机器故障请求来了会把它抢占
            with repairman.request(priority=2) as req:
                yield req                # 等待获得维修工                start = env.now
                try:
                    yield env.timeout(done_in)  # 执行杂务
                    done_in = 0                 # 正常完成
                except simpy.Interrupt:
                    # 被机器维修请求抢占，算剩余时间
                    done_in -= env.now - start
                    # 退出 with 块，自动释放资源
                    # 外层 while 重新排队请求维修工

这里有个微妙的地方：with repairman.request() 放在内层 while 里面，而不是外面。这样每次被抢占后，杂务进程会重新排队申请维修工，而不是一直占着资源等。这是 SimPy 处理"可中断任务"的标准写法。

---

主程序：把所有东西组装起来

print('Machine shop')random.seed(RANDOM_SEED)             # 固定随机种子#  创建仿真环境，这是一切的起点
env = simpy.Environment()#  创建抢占式资源：维修工，容量为 1（同时只能服务一个请求）
repairman = simpy.PreemptiveResource(env, capacity=1)# 创建 10 台机器，每台机器的 __init__ 会自动启动它的两个进程
machines = [Machine(env, f'Machine {i}', repairman) for i in range(NUM_MACHINES)]# 把维修工杂务进程也注册进来
env.process(other_jobs(env, repairman))#  启动仿真，跑到 SIM_TIME（40320 分钟）为止
env.run(until=SIM_TIME)# 输出结果
print(f'Machine shop results after {WEEKS} weeks')
for machine in machines:
    print(f'{machine.name} made {machine.parts_made} parts.')

env.run(until=SIM_TIME) 这一行是真正的发动机。它驱动所有已注册的进程按时间顺序推进，直到仿真时钟走完 40320 分钟。

---

三个机制，一张图说清楚

进程与 yield 的协作

working 进程                      仿真调度器
    │                                 │
    │── yield timeout(done_in) ──▶    │  记录事件，推进时钟
    │                                 │
    │◀── 时间到，恢复执行 ─────────────│
    │                                 │
    │── yield req（等资源）──────────▶ │  资源忙则挂起
    │◀── 资源可用，恢复执行 ───────────│

中断的传递路径

break_machine 进程              working 进程
      │                              │
      │  time_to_failure() 到期      │  正在 yield timeout(done_in)
      │                              │
      │── process.interrupt() ──▶   │  timeout 被取消
      │                              │  抛出 simpy.Interrupt
      │                              │  进入 except 块处理故障

抢占资源的优先级调度

时间线：
  维修工做杂务（priority=2）─────────────────▶
                      │
                      │  机器故障请求（priority=1）到来
                      ↓
  杂务被中断，维修工转去修机器（priority=1）──▶
                                    │
                                    │  修完
                                    ↓
  维修工回来继续做杂务（剩余时间）────────────▶

---

读完这个例子，你真正学到了什么

通过这个机器车间案例，我们深入理解了SimPy如何利用生成器实现并发模拟，掌握进程调度、资源抢占和中断处理三大核心机制。这种基于时间推进的仿真方法，为复杂系统建模提供了优雅的解决方案。