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

【问题标题】：How can I implement a multi-producer, multi-consumer paradigm in Gevent?如何在 Gevent 中实现多生产者、多消费者范式？
【发布时间】：2012-03-01 21:46:52
【问题描述】：

我有一些依赖 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 循环，则脚本运行良好。

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

有什么想法吗？

谢谢

【问题讨论】：

标签： python gevent

【解决方案1】：

我认为q.join() 在将任何内容放入队列并立即退出之前会这样做。在加入队列之前尝试加入所有生产者。

【讨论】：

嗨，Zch，我没有完全听从你的回答。你能贴一个小的sn-p吗？这会澄清一些事情。
@MridangAgarwalla - 在q.join() 之前写for producer in producers: producer.join()。这样你首先等到所有生产者结束他们的工作，然后直到队列为空。
啊哈，也许我实施错了。我希望我的生产者和消费者同时运行，即生产者不断添加到队列中直到他们完成，而消费者提供它直到所有队列项完成并且生产者不再向队列中添加东西。
@MridangAgarwalla - 只有主线程会等待，其他人会做他们的工作。更准确地说：它会一直阻塞，直到所有生产者结束他们的工作并且消费者宣布q.task_done() 与添加到队列中的东西一样多。

【解决方案2】：

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

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

# 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.

【讨论】：

【解决方案3】：

您要做的是在生产者和工作人员通信时阻止主程序。阻塞队列将等到队列为空，然后屈服，这可能是立即的。将其放在程序的末尾而不是 q.join()

gevent.joinall(producers)

【讨论】：

【解决方案4】：

我遇到了和你一样的问题。您的代码的主要问题是您的生产者已在 gevent 线程中产生，这使工作人员无法立即获得任务。

我建议你应该在主进程中运行producer()，而不是在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)


producer()

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

上面的代码很有意义.. :)

【讨论】：