python - 如何在 Gevent 中实现多生产者、多消费者范式？

Question

我有一些依赖于 I/O 重度阻塞调用的生产者函数和一些也依赖于 I/O 重度阻塞调用的消费者函数。为了加快它们的速度，我使用 Gevent 微线程库作为胶水。

这是我的范例:

import gevent
from gevent.queue import *
import time
import random

q = JoinableQueue()
workers = []
producers = []

def do_work(wid, value):
    gevent.sleep(random.randint(0,2))
    print 'Task', value, 'done', wid

def worker(wid):
    while True:
        item = q.get()
        try:
            print "Got item %s" % item
            do_work(wid, item)
        finally:
            print "No more items"
            q.task_done()


def producer():
    while True:
        item = random.randint(1, 11)
        if item == 10:
            print "Signal Received"
            return
        else:
            print "Added item %s" % item
            q.put(item)



for i in range(4):
    workers.append(gevent.spawn(worker, random.randint(1, 100000)))

#This doesnt work.
for j in range(2):
    producers.append(gevent.spawn(producer))

#Uncommenting this makes this script work.
#producer()

q.join()

我有四个消费者，想有两个生产者。生产者在发出信号时退出，即 10。消费者继续为这个队列提供数据，当生产者和消费者结束时，整个任务结束。

但是，这是行不通的。如果我注释掉产生多个生产者的 for 循环并只使用一个生产者，脚本运行正常。

我似乎无法弄清楚我做错了什么。

有什么想法吗？

谢谢

Answer 1

当队列没有未完成的工作时，您实际上并不想退出，因为从概念上讲，那不是应用程序应该完成的时间。

您想在制作人完成后退出，然后没有未完成的工作。

# Wait for all producers to finish producing
gevent.joinall(producers)
# *Now* we want to make sure there's no unfinished work
q.join()
# We don't care about workers. We weren't paying them anything, anyways
gevent.killall(workers)
# And, we're done.