python - gevent - 将一个值并行传递给许多 greenlets

Question

我正在尝试实现一个简单的 gevent 设置。有一个发件人应该并行向多个服务员发送一个值。 Event 类最接近解决此问题，如下所示。

每三秒，setter 创建一个事件，解除对所有等待者的阻塞。事件立即被清除，所以服务员再次阻塞直到下一次。

import gevent
from gevent.event import Event

evt = Event()

def setter():
    '''After 3 seconds, wake all threads waiting on the value of evt'''
    while True:
        gevent.sleep(3)
        evt.set()
        evt.clear()

def waiter(arg):
    while True:
        evt.wait()
        print("waiter {}".format(arg))

def main():
    gevent.joinall([
        gevent.spawn(setter),
        gevent.spawn(waiter,1),
        gevent.spawn(waiter,2),
        gevent.spawn(waiter,3),
    ])

if __name__ == '__main__': main()

现在，我只需要执行此操作，并向服务员传递一个值。显而易见的选择是使用 AsyncResult。然而，清除 AsyncResult 对象是不可能的，所以等待者最终陷入无限循环。

您有什么想法可以实现吗？

Answer 1

我认为最好的办法是为此使用队列。我创建了一个 BroadcastQueue 类，它可以更轻松地管理向多个消费者发送一个值。生产者调用 BroadcastQueue.broadcast()，这将向所有已注册的消费者发送一个值。消费者通过调用 BroadcastQueue.register 进行注册，它返回一个唯一的 gevent.queue.Queue() 对象。然后，消费者使用该对象从生产者那里获取消息。

import gevent
from gevent.queue import Queue


class BroadcastQueue(object):
    def __init__(self):
        self._queues = []

    def register(self):
        q = Queue()
        self._queues.append(q)
        return q

    def broadcast(self, val):
        for q in self._queues:
            q.put(val)


def setter(bqueue):
    '''After 3 seconds, wake all threads waiting on the value of evt'''
    while True:
        gevent.sleep(3)
        bqueue.broadcast("hi")

def waiter(arg, bqueue):
    queue = bqueue.register()
    while True:
        val = queue.get()
        print("waiter {} {}".format(arg, val))

def main():
    bqueue = BroadcastQueue()
    gevent.joinall([
        gevent.spawn(setter, bqueue),
        gevent.spawn(waiter, 1, bqueue),
        gevent.spawn(waiter, 2, bqueue),
        gevent.spawn(waiter, 3, bqueue),
    ])

if __name__ == '__main__':
    main()

输出:

waiter 1 hi
waiter 2 hi
waiter 3 hi
waiter 1 hi
waiter 2 hi
waiter 3 hi
waiter 1 hi
waiter 2 hi
waiter 3 hi
waiter 1 hi
waiter 2 hi
waiter 3 hi
waiter 1 hi
waiter 2 hi
waiter 3 hi