如何在SimPy中实现满员触发的批量资源（Batch Resource）

我明白你的需求啦——你希望渡轮要等凑够满员之后，再统一出发运输，而不是先让乘客单独“等待”再走。咱们来看看原代码里的问题，然后一步步调整：

原代码的问题分析 #

你当前的person函数里，请求渡轮后先执行了一段1单位的timeout作为“等待”，但这段时间其实和渡轮的批量运输逻辑脱节了。而且BatchResource的request处理里，虽然攒够批次会启动运输，但释放请求的时机是运输完成后，可乘客那边已经先走完了“等待”流程，导致橙色块只显示了单独的等待时间，没把渡轮运输的时间包含进去。

修改后的实现方案 #

我们需要调整两个核心部分：

• 让BatchResource在攒够批次后，先完成运输流程，再统一释放所有请求的乘客
• 调整person函数里的逻辑，让乘客的“渡轮阶段”直接对应渡轮攒人+运输的完整时间

下面是修改后的完整代码：

import simpy
import matplotlib.pyplot as plt


class Person:
    def __init__(self, name):
        self.name = name
        self.timeline = []

    def log(self, event, time):
        self.timeline.append((event, time))


class BatchResource(simpy.Resource):
    def __init__(self, env, capacity, batch_size, process_time):
        super().__init__(env, capacity)
        self.batch_size = batch_size
        self.process_time = process_time
        self.batch_queue = []

    def request(self):
        req = super().request()
        self.batch_queue.append(req)

        # 只有当队列里的请求数达到批量大小，才启动运输
        if len(self.batch_queue) >= self.batch_size:
            self._env.process(self.process_batch())

        return req

    def process_batch(self):
        # 取出当前批次的请求
        batch = self.batch_queue[:self.batch_size]
        self.batch_queue = self.batch_queue[self.batch_size:]

        print(f"[{self._env.now}] 渡轮满员！载着{len(batch)}人出发。")
        # 模拟渡轮过河的时间
        yield self._env.timeout(self.process_time)
        print(f"[{self._env.now}] 渡轮已抵达对岸。")

        # 批量释放所有乘客的请求
        for req in batch:
            if not req.triggered:
                req.succeed()


def person(env, person, ferry, casses_before, casses_after):
    # 过河前的流程
    with casses_before.request() as request:
        yield request
        person.log("start_before", env.now)
        yield env.timeout(1)
        person.log("end_before", env.now)

    # 等待渡轮攒满并完成运输——这里的等待直接对应渡轮的整个流程
    person.log("start_wait", env.now)
    with ferry.request() as request:
        yield request
        person.log("end_wait", env.now)

    # 过河后的流程
    with casses_after.request() as request:
        yield request
        person.log("start_after", env.now)
        yield env.timeout(1)
        person.log("end_after", env.now)


def setup(env, num_people, capacity, people):
    casses_before = simpy.Resource(env, capacity=3)
    casses_after = simpy.Resource(env, capacity=3)
    ferry = BatchResource(env, capacity=capacity, batch_size=capacity, process_time=5)

    for i in range(num_people):
        person_instance = Person(f"Person {i}")
        people.append(person_instance)
        env.process(person(env, person_instance, ferry, casses_before, casses_after))
        yield env.timeout(0.5)  # 每0.5时间单位来一个人


# 启动模拟
env = simpy.Environment()
people = []
env.process(setup(env, num_people=30, capacity=10, people=people))
env.run(until=50)

# 绘制甘特图
fig, ax = plt.subplots(figsize=(10, 8))

colors = {"before": "tab:blue", "wait": "tab:orange", "after": "tab:green"}

for i, person in enumerate(people):
    start_time = None
    stage = None
    for event, time in person.timeline:
        if "start" in event:
            start_time = time
            stage = event.split("_")[1]
        elif "end" in event:
            end_time = time
            ax.broken_barh(
                [(start_time, end_time - start_time)],
                (i * 10, 9),
                facecolors=(colors[stage]),
            )

ax.set_yticks([i * 10 + 5 for i in range(len(people))])
ax.set_yticklabels([person.name for person in people])
ax.set_xlabel("时间")
ax.set_ylabel("乘客")
ax.grid(True)

plt.show()

关键修改点说明 #

• 调整person函数的日志时机：把start_wait的日志移到请求渡轮之前，这样橙色块会从乘客开始等待渡轮的时间算起，直到渡轮运输完成后才结束，完美对应“攒人+运输”的整个过程。
• 移除了原代码中请求渡轮后的多余timeout：原代码里的yield env.timeout(1)是多余的，现在乘客的等待时间完全由渡轮的批量逻辑决定。
• 保持BatchResource的核心逻辑：渡轮依然会等凑够batch_size的乘客才启动运输，运输完成后统一释放所有乘客的请求，这样所有同批次的乘客会同时完成渡轮阶段。

这样修改后，甘特图里的橙色块就会准确显示乘客从开始等渡轮，到渡轮满员出发再抵达对岸的完整时间啦！

备注：内容来源于stack exchange，提问作者Alexander Lyapin