Simpy多缓冲区系统中单缓冲区阻塞且不影响整体流程的实现问题

核心逻辑是为每个缓冲区单独维护独立的可用状态，不要在全局主流程中调用env.timeout，而是将冷却逻辑绑定到单个缓冲区实例上，和系统其他进程隔离。

具体实现思路 #

• 自定义缓冲区类，封装容量、存储资源、是否允许取出的标记三个核心属性
• 每次从缓冲区取出产品后，单独启动一个SimPy进程执行15分钟的冷却计时，计时结束后再修改该缓冲区的可取标记，这个独立进程的timeout不会阻塞系统其他调度
• 取出操作前先判断当前缓冲区的可取标记，若处于冷却中则让取出请求排队等待，冷却结束后自动触发排队的取出逻辑，存入操作不受该标记限制

最小可运行示例代码 #

import simpy

class CoolingBuffer:
    def __init__(self, env, capacity=1, cool_down=15):
        self.env = env
        self.store = simpy.Store(env, capacity=capacity)
        self.can_take = True  # 标记当前缓冲区是否允许取出
        self.cool_down = cool_down
        self.take_queue = simpy.Store(env)  # 缓存待处理的取出请求

    def put(self, item):
        # 存入操作不受冷却限制，直接存入存储区
        return self.store.put(item)

    def _cool_down_process(self):
        # 独立的冷却进程，仅影响当前缓冲区
        self.can_take = False
        yield self.env.timeout(self.cool_down)
        self.can_take = True
        # 冷却结束后处理队列中等待的取出请求
        if len(self.take_queue.items) > 0 and len(self.store.items) > 0:
            req = yield self.take_queue.get()
            item = yield self.store.get()
            req.succeed(item)
            # 取出后再次触发冷却
            self.env.process(self._cool_down_process())

    def take(self):
        # 取出请求处理逻辑
        req = self.env.event()
        if self.can_take and len(self.store.items) > 0:
            # 可直接取出
            item = yield self.store.get()
            req.succeed(item)
            self.env.process(self._cool_down_process())
        else:
            # 不可取出则加入请求队列
            self.take_queue.put(req)
        return req

# 测试用例
def test_system(env, buffers):
    # 先给4个缓冲区都存入产品
    for i, buf in enumerate(buffers):
        yield buf.put(f"product_{i}")
    # 每个缓冲区各取一次
    for buf in buffers:
        item = yield buf.take()
        print(f"取出 {item}，时间：{env.now}")
    # 10分钟后尝试再次取出（此时缓冲区还在冷却）
    yield env.timeout(10)
    print(f"尝试再次取出，时间：{env.now}")
    for buf in buffers:
        def take_task(b):
            item = yield b.take()
            print(f"冷却结束后取出 {item}，时间：{env.now}")
        env.process(take_task(buf))
    # 运行到30分钟看完整流程
    yield env.timeout(10)

if __name__ == "__main__":
    env = simpy.Environment()
    buffers = [CoolingBuffer(env) for _ in range(4)]
    env.process(test_system(env, buffers))
    env.run(until=30)

原理解释 #

你之前出现全局暂停的核心原因是把env.timeout(15)放在了和系统主逻辑同属的进程中，SimPy的单进程异步调度特性会让当前进程阻塞在timeout上，导致后续所有逻辑都被暂停。而上述方案中每个缓冲区的冷却逻辑都在独立的进程里执行，timeout只会阻塞当前缓冲区的冷却进程，完全不影响其他缓冲区和系统主逻辑的调度。

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