python – 需要一个线程安全的异步消息队列

我正在寻找一个Python类(最好是标准语言的一部分,而不是第三方库)来管理异步“广播风格”消息.

我将有一个线程将消息放入队列(‘putMessageOnQueue’方法不能阻止),然后多个其他线程将等待消息,可能会调用一些阻塞’waitForMessage’函数.当消息放在队列上时,我希望每个等待的线程都获得它自己的消息副本.

我查看了内置的Queue类,但我不认为这是合适的,因为消费消息似乎涉及将它们从队列中删除,因此只有1个客户端线程会看到每个.

这似乎应该是一个常见的用例,任何人都可以推荐一个解决方案吗？

解决方法:

我认为这种方法的典型方法是为每个线程使用单独的消息队列,并将消息推送到之前已注册过接收此类消息的每个队列.

像这样的东西应该工作,但它是未经测试的代码……

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()

标签：python,multithreading,message-queue,queue
来源： https://codeday.me/bug/20191007/1868622.html