SimPy租赁服务模拟未完成所有进程yield语句即终止问题排查

核心问题在于逐个获取资源的逻辑导致死锁：当多个进程在同一时间点请求的资源总量超过Store剩余容量时，每个进程会先拿走部分资源，但没有任何进程能凑齐所需的全部数量，进而无法归还资源，所有进程陷入无限等待。

示例场景：Store容量为4，3个进程同时各请求4个电池。每个进程先拿走1个后，Store剩余1个，此时每个进程还缺3个，没有进程能继续获取足够资源，模拟彻底卡住。

核心思路：让进程先等待直到Store中有足够数量的资源，再一次性获取全部所需资源，避免部分拿取引发的死锁。

修改后的完整代码如下：

import simpy
import random
from numpy.random import default_rng

rng = default_rng()

# 模拟天数及每天的用例数量
arrivals = [3, 3, 3]

# 模拟总时长
runtime = 200

# 电池可用数量
availability = 4

# 用例配置列表
list_of_uscases = [
    {
        "usecase": "eMotorbike",
        "dep_mean": 7,
        "dep_std": 1,
        "num_Batteries": 4
    },
    {
        "usecase": "eCar",
        "dep_mean": 17,
        "dep_std": 1,
        "num_Batteries": 4
    },
    {
        "usecase": "eBike",
        "dep_mean": 12,
        "dep_std": 1,
        "num_Batteries": 4
    },
]

# 生成租赁进程的函数
def usecase_gen(env):
    day = 0
    for daily_count in arrivals:
        count = 0
        for _ in range(daily_count):
            usecase = random.choice(list_of_uscases)
            env.process(job(env, count, day, usecase["num_Batteries"], usecase))
            count += 1
        day += 1
        yield env.timeout(24)  # 模拟一天24小时

# 单个租赁进程逻辑
def job(env, count, day, resources_required, usecase):
    yield env.timeout(count)  # 用例触发的时间偏移
    resources_used = []

    print(f'第{day}天 用例{usecase["usecase"]}_{count} 需要{resources_required}个电池，当前时间：{env.now}')

    # 等待直到Store中有足够的电池
    yield env.process(wait_for_enough_resources(env, resources, resources_required))
    
    # 一次性获取所需全部电池
    for _ in range(resources_required):
        battery = yield resources.get()
        resources_used.append(battery)

    print(f'第{day}天 用例{usecase["usecase"]}_{count} 成功租赁{resources_required}个电池，当前时间：{env.now}')

    # 模拟电池使用时长
    usagetime = 8
    yield env.timeout(usagetime)

    # 归还电池
    for battery in resources_used:
        yield resources.put(battery)

    print('-----------------------------------------')
    print(f'用例{usecase["usecase"]}_{count}（第{day}天租赁）使用了{resources_required}个电池，时长{usagetime}小时')
    print(f'用例{usecase["usecase"]}_{count}（第{day}天租赁）归还{resources_required}个电池，当前时间：{env.now}')
    print('-----------------------------------------')

# 辅助函数：等待Store中资源数量满足需求
def wait_for_enough_resources(env, store, required):
    while len(store.items) < required:
        # 监听Store的放入/取出事件，避免无限等待
        yield store.put_event | store.get_event

# 初始化模拟环境
env = simpy.Environment()

# 创建电池Store
resources = simpy.Store(env, capacity=availability)

# 填充初始电池
for i in range(availability):
    resources.put({'id': i})

# 启动用例生成进程
env.process(usecase_gen(env))

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

1. 新增资源等待逻辑：添加wait_for_enough_resources辅助函数，循环监听Store的put_event和get_event，直到Store内的电池数量满足进程需求。
2. 批量获取资源：进程先等待足够资源，再一次性拿取全部所需电池，避免部分拿取导致的死锁。
3. 优化输出格式：使用f-string简化打印语句，提升可读性。

当availability=4时，第一个进程会在时间0拿到4个电池，使用8小时后在时间8归还；第二个进程在时间0进入等待，直到时间8电池归还后完成租赁，所有进程均能正常完成「租赁-使用-归还」流程，不会出现死锁。

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