Python Simpy如何实现任一yield即终止的并发生成器循环以建模设备维保流程

核心实现逻辑 #

你要的效果可以通过Simpy内置的事件等待机制实现，核心逻辑是：

• 每次同时启动三类停机触发进程，用AnyOf监听所有进程的第一个返回结果
• 拿到第一个触发的事件后，主动终止其余两个未触发的进程
• 根据触发事件的类型执行对应维保流程，完成后重置所有进程重新启动循环

import simpy
import random

# 基础参数配置
MTBF = 150  # 平均无故障工作时间，单位：小时
# 定时服务间隔支持多档配置，可替换为[4*7*24, 13*7*24, 26*7*24, 52*7*24]对应4/13/26/52周需求
SERVICE_INTERVAL = 600  # 定时服务间隔，单位：小时
INSPECT_REPAIR_INTERVAL = 300  # 例行检查间隔，单位：小时

# 三类停机触发进程，返回对应事件类型标识
def run_to_failure(env):
    # 生成指数分布的故障间隔时长
    failed_after = random.expovariate(1/MTBF)
    yield env.timeout(failed_after)
    return "故障停机"

def run_to_service(env):
    yield env.timeout(SERVICE_INTERVAL)
    return "定时服务到期停机"

def run_to_repair(env):
    yield env.timeout(INSPECT_REPAIR_INTERVAL)
    return "例行检查及后续修复停机"

# 主控制流程
def maintenance_control(env):
    while True:
        # 同时启动三类触发进程
        failure_proc = env.process(run_to_failure(env))
        service_proc = env.process(run_to_service(env))
        inspect_proc = env.process(run_to_repair(env))
        
        # 等待任意一个进程先完成
        completed = yield simpy.AnyOf(env, [failure_proc, service_proc, inspect_proc])
        # 拿到第一个触发的事件类型
        trigger_type = list(completed.values())[0]
        
        # 终止剩下两个未触发的进程，避免继续运行
        for proc in [failure_proc, service_proc, inspect_proc]:
            if not proc.triggered:
                proc.interrupt()
        
        # 此处为示例输出，可替换为自定义业务逻辑
        print(f"[当前时间 {env.now:.2f} 小时] 触发停机：{trigger_type}，开始执行对应维保流程")
        
        # 可自定义不同类型维保的耗时逻辑，示例为固定耗时
        if trigger_type == "故障停机":
            yield env.timeout(2)
        elif trigger_type == "定时服务到期停机":
            yield env.timeout(4)
        elif trigger_type == "例行检查及后续修复停机":
            yield env.timeout(3)
        
        print(f"[当前时间 {env.now:.2f} 小时] {trigger_type} 维保完成，设备恢复运行，重置所有计时规则")

# 启动仿真
env = simpy.Environment()
env.process(maintenance_control(env))
env.run(until=5000)  # 仿真运行到5000小时停止，可自行修改终止条件

代码说明 #

• 你原来写的while True不需要放在三个子进程里，主控制流程的循环已经实现了每次维保后重置所有计时的逻辑，完全符合“服务/检查完成后故障概率分布重置”的要求
• 如果要替换定时服务的多档周期间隔，只需要调整run_to_service的逻辑即可，示例如下：

  def run_to_service(env):
      # 按规则选择4、13、26、52周对应时长
      service_intervals = [4*7*24, 13*7*24, 26*7*24, 52*7*24]
      selected_interval = random.choice(service_intervals) # 也可替换为自定义调度规则
      yield env.timeout(selected_interval)
      return "定时服务到期停机"
  

• 后续的业务逻辑可以直接在主流程的触发判断分支里扩展，也可以单独封装成函数调用。

内容的提问来源于stack exchange，提问作者Joseph Mwanza