Python中如何捕获SimPy进程内抛出的异常？

你遇到的核心问题是：SimPy进程是异步调度的——env.process()启动的diff_process会被加入SimPy的事件队列等待调度，和调用它的go()/function_call()不在同一个同步执行流里，因此go()里的try-except完全触及不到异步进程抛出的异常。

方案1：等待SimPy进程完成并捕获异常 #

如果go()可以改造为SimPy进程，直接通过yield等待diff_process执行完成，就能在yield语句处捕获异常；如果go()必须是普通函数，也可以保存Process对象，后续检查其exception属性。

改造为SimPy进程的实现（推荐） #

import simpy

# 项目自定义异常类
class CustomSimulationError(Exception):
    """仿真流程中抛出的自定义异常"""
    pass

env = simpy.Environment()

def go():
    try:
        # 等待diff_process执行完成，异常会在此处抛出
        yield function_call()
    except CustomSimulationError as e:
        print(f"捕获到自定义异常: {str(e)}")
        # 这里添加你的异常处理逻辑（比如日志记录、流程回滚等）

def function_call():
    # 返回Process对象，让go可以yield等待
    return env.process(diff_process())

def diff_process():
    yield env.timeout(1)
    # 抛出自定义异常
    raise CustomSimulationError("diff_process执行超时后触发错误")

def helper():
    yield go()  # 等待go进程完成
    yield env.timeout(2)

def main():
    env.process(helper())
    env.run()

if __name__ == "__main__":
    main()

普通函数场景的实现 #

如果go()不能作为SimPy进程，可保存Process对象，在后续的SimPy流程中检查异常：

import simpy

class CustomSimulationError(Exception):
    pass

env = simpy.Environment()
diff_proc = None

def go():
    global diff_proc
    diff_proc = function_call()

def function_call():
    return env.process(diff_process())

def diff_process():
    yield env.timeout(1)
    raise CustomSimulationError("diff_process执行出错")

def check_exception():
    yield env.timeout(1.5)  # 等待diff_process执行完成
    if diff_proc.exception:
        print(f"捕获到异常: {diff_proc.exception}")
        # 执行异常处理逻辑

def helper():
    go()
    yield check_exception()
    yield env.timeout(2)

def main():
    env.process(helper())
    env.run()

if __name__ == "__main__":
    main()

方案2：使用事件中断机制 #

如果需要在diff_process抛出异常时中断其他SimPy进程，可结合SimPy的Interrupt机制实现：

import simpy

class CustomSimulationError(Exception):
    pass

env = simpy.Environment()

def go(target_proc):
    try:
        yield target_proc  # 等待目标进程完成
    except simpy.Interrupt as interrupt:
        if isinstance(interrupt.cause, CustomSimulationError):
            print(f"通过中断捕获到异常: {str(interrupt.cause)}")
            # 处理中断后的逻辑（比如终止当前流程、触发告警）

def diff_process(interrupt_target):
    try:
        yield env.timeout(1)
        raise CustomSimulationError("diff_process执行出错")
    except CustomSimulationError as e:
        # 中断指定的目标进程，并将异常作为中断原因传递
        interrupt_target.interrupt(e)

def helper():
    # 创建一个长期运行的目标进程，让go等待它
    target_proc = env.process(env.timeout(10))
    # 启动go进程等待目标进程
    env.process(go(target_proc))
    # 启动diff_process，传入要中断的目标进程
    env.process(diff_process(target_proc))
    yield env.timeout(5)

def main():
    env.process(helper())
    env.run()

if __name__ == "__main__":
    main()

• SimPy的异常捕获必须在SimPy事件循环的执行流内完成，同步代码的try-except无法捕获异步调度的进程异常。
• 自定义异常类的使用和原生异常完全一致，只需在raise和except处替换为你的自定义类即可。
• 若无需联动其他进程，优先选择方案1的yield等待方式，逻辑更直观；若需要触发其他进程的中断，再使用方案2的事件中断机制。

内容的提问来源于stack exchange，提问作者Charles Harlem