多个消费者，是否可以克隆队列（gevent）？

Question

我想要做这样的事情（1个队列，并在多个用户）：

import gevent 
from gevent import queue 

q=queue.Queue() 
q.put(1) 
q.put(2) 
q.put(3) 
q.put(StopIteration) 

def consumer(qq): 
    for i in qq: 
     print i 

jobs=[gevent.spawn(consumer,i) for i in [q,q]] 

gevent.joinall(jobs) 

但是这是不可能的......队列由JOB1消耗...所以作业2将永远阻塞。 它给了我例外gevent.hub.LoopExit: This operation would block forever。

我会每个消费者将能够从开始消耗完整的队列。 （应该显示1,2,3,1,2,3或1,1,2,2,3,3 ...从不知道）

一个想法应该是在产卵前克隆队列，但这是不可能的使用复制（浅/深）模块;-(

是否有另一种方式来做到这一点？

[编辑] 你觉得什么？

import gevent 
from gevent import queue 

class MasterQueueClonable(queue.Queue): 
    def __init__(self,*a,**k): 
     queue.Queue.__init__(self,*a,**k) 

     self.__cloned = [] 
     self.__old=[] 

    #override 
    def get(self,*a,**k): 
     e=queue.Queue.get(self,*a,**k) 
     for i in self.__cloned: i.put(e) # serve to current clones 
     self.__old.append(e)    # save old element 
     return e 

    def clone(self): 
     q=queue.Queue() 
     for i in self.__old: q.put(i) # feed a queue with elements which are out 
     self.__cloned.append(q)   # stock the queue, to be able to put newer elements too 
     return q 

q=MasterQueueClonable() 
q.put(1) 
q.put(2) 
q.put(3) 
q.put(StopIteration) 

def consumer(qq): 
    for i in qq: 
     print id(qq),i 

jobs=[gevent.spawn(consumer,i) for i in [q.clone(), q ,q.clone(),q.clone()]] 
gevent.joinall(jobs) 

它是基于这样的理念RyanYe。有一个没有调度员的“主队列”。 我的主队列覆盖GET方法，并可以派发到按需克隆。 此外，可以在开始masterqueue之后创建“克隆”（使用__old技巧）。

Answer 1

我建议你创建一个greenlet将工作分发给消费者。示例代码：

import gevent 
from gevent import queue 

master_queue=queue.Queue() 
master_queue.put(1) 
master_queue.put(2) 
master_queue.put(3) 
master_queue.put(StopIteration) 

total_consumers = 10 
consumer_queues = [queue.Queue() for i in xrange(total_consumers)] 

def dispatcher(master_queue, consumer_queues): 
    for i in master_queue: 
     [j.put(i) for j in consumer_queues] 
    [j.put(StopIteration) for j in consumer_queues] 

def consumer(qq): 
    for i in qq: 
     print i 

jobs=[gevent.spawn(dispatcher, q, consumer_queues)] + [gevent.spawn(consumer, i) for i in consumer_queues] 
gevent.joinall(jobs) 

更新：修复了消费者队列丢失的StopIteration。谢谢你指出。

Answer 2

在由Ryan烨一行答案是错过了在调度（）函数的末尾： [j.put（StopIteration异常），用于consumer_queues J] 没有它，我们仍然可以得到“gevent.hub.LoopExit：这个操作会永远阻塞'，因为'我在master_queue'循环不会将StopIteration异常复制到consumer_queues中。

（对不起，我不能发表评论还没有，所以我把它写成一个separete答案。）

Answer 3

我已经addedcopy()方法队列类：

>>> import gevent.queue 
>>> q = gevent.queue.Queue() 
>>> q.put(5) 
>>> q.copy().get() 
5 
>>> q 
<Queue at 0x1062760d0 queue=deque([5])> 

让我知道，如果它帮助。