需要一个线程安全的异步消息队列 [英] Need a thread-safe asynchronous message queue

问题描述

我正在寻找一个Python类(最好是标准语言的一部分，而不是第3方库)来管理异步广播样式"消息.

I'm looking for a Python class (preferably part of the standard language, rather than a 3rd party library) to manage asynchronous 'broadcast style' messaging.

我将有一个线程将消息放入队列("putMessageOnQueue"方法不得阻塞)，然后有多个其他线程都将在等待消息，大概调用了一些阻塞的"waitForMessage"函数.当消息放入队列时，我希望每个等待的线程都获得自己的消息副本.

I will have one thread which puts messages on the queue (the 'putMessageOnQueue' method must not block) and then multiple other threads which will all be waiting for messages, having presumably called some blocking 'waitForMessage' function. When a message is placed on the queue I want each of the waiting threads to get its own copy of the message.

我已经看过内置的Queue类，但是我认为这不合适，因为使用消息似乎涉及从队列中删除消息，因此每个消息只有一个客户端线程.

I've looked at the built-in Queue class, but I don't think this is suitable because consuming messages seems to involve removing them from the queue, so only 1 client thread would see each one.

这似乎应该是一个普通的用例，任何人都可以推荐解决方案吗?

This seems like it should be a common use-case, can anyone recommend a solution?

推荐答案

我认为典型的解决方法是为每个线程使用一个单独的消息队列，并将消息推送到先前已注册接收消息的每个队列中这样的消息.

I think the typical approach to this is to use a separate message queue for each thread, and push the message onto every queue which has previously registered an interest in receiving such messages.

像这样的事情应该起作用，但这是未经测试的代码...

Something like this ought to work, but it's untested code...

from time import sleep
from threading import Thread
from Queue import Queue

class DispatcherThread(Thread):

   def __init__(self, *args, **kwargs):
       super(DispatcherThread, self).__init__(*args, **kwargs)
       self.interested_threads = []

   def run(self):
       while 1:
           if some_condition:
               self.dispatch_message(some_message)
           else:
               sleep(0.1)

   def register_interest(self, thread):
       self.interested_threads.append(thread)

   def dispatch_message(self, message):
       for thread in self.interested_threads:
           thread.put_message(message)



class WorkerThread(Thread):

   def __init__(self, *args, **kwargs):
       super(WorkerThread, self).__init__(*args, **kwargs)
       self.queue = Queue()


   def run(self):

       # Tell the dispatcher thread we want messages
       dispatcher_thread.register_interest(self)

       while 1:
           # Wait for next message
           message = self.queue.get()

           # Process message
           # ...

   def put_message(self, message):
       self.queue.put(message)


dispatcher_thread = DispatcherThread()
dispatcher_thread.start()

worker_threads = []
for i in range(10):
    worker_thread = WorkerThread()
    worker_thread.start()
    worker_threads.append(worker_thread)

dispatcher_thread.join()

这篇关于需要一个线程安全的异步消息队列的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！