1

我正在尝试创建一个生产者/消费者模拟，消费者在其中批量处理项目。问题是 Store.get() 函数一被调用就会从 Store 中删除项目，但我需要它等到我调用了 yield：

import simpy

def producer(env, Q):
    item = 0
    while True:
        yield Q.put(item)
        print('Submit item:%d'%item)
        item += 1

def consumer(env, Q):
    while True:
        yield env.timeout(20)
        events = [Q.get() for i in range(4)]
        items = yield env.all_of(events)
        print([items for items in items.values()])


env = simpy.Environment()
maxQD = 2
Q = simpy.Store(env, capacity=maxQD)

env.process(producer(env, Q))
env.process(consumer(env, Q))
env.run(until=500)

产生以下输出：

Submit item:0
Submit item:1
Submit item:2
Submit item:3
Submit item:4
[0, 1, 2, 3]
Submit item:5
Submit item:6
Submit item:7
Submit item:8
[4, 5, 6, 7]
Submit item:9
Submit item:10
Submit item:11
Submit item:12
[8, 9, 10, 11]
...

将 maxQD 设置为 2 时，我会期望：

Submit item:0
Submit item:1

消费者阻塞直到成功获得 4 个项目，而生产者无法添加超过 2 个。

你可以通过检查 len(Q.items) 来解决这个问题：

import simpy

def producer(env, Q):
    item = 0
    while True:
        yield Q.put(item)
        print('Submit item:%d'%item)
        item += 1

def consumer(env, Q):
    while True:
        yield env.timeout(20)
        if len(Q.items) >= 4:
            events = [Q.get() for i in range(4)]
            items = yield env.all_of(events)
            print([items for items in items.values()])


env = simpy.Environment()
maxQD = 4
Q = simpy.Store(env, capacity=maxQD)

env.process(producer(env, Q))
env.process(consumer(env, Q))
env.run(until=500)

但是你仍然会得到令人沮丧的行为，即 get() 在 yield 之前删除了项目，这使得它看起来像 5 个项目被添加到 Q （注意 maxQD 更改为 4）：

Submit item:0
Submit item:1
Submit item:2
Submit item:3
Submit item:4
[0, 1, 2, 3]
4

1 回答 1

2

通过子类化 Store 解决：

class BatchGet(simpy.resources.base.Get):
    def __init__(self, resource, count):
        self.count = count
        super().__init__(resource)

class BatchStore(simpy.resources.store.Store):
    get = simpy.core.BoundClass(BatchGet)

    def _do_put(self, event):
        if len(self.items) + len(event.item) <= self._capacity:
            self.items.extend(event.item)
            event.succeed()   

    def _do_get(self, event):
        count = event.count
        if len(self.items) >= count:
            ret = self.items[:count]
            self.items = self.items[count:]
            event.succeed(ret)

Puts 必须获取一个列表（因为它是一批项目），并返回一个列表。

于 2016-07-01T17:37:25.113 回答