我正在寻找一个 Python 类（最好是标准语言的一部分，而不是第 3 方库）来管理异步“广播风格”消息传递。
我将有一个线程将消息放入队列（'putMessageOnQueue' 方法不得阻塞），然后有多个其他线程都将等待消息，大概调用了一些阻塞'waitForMessage' 函数。当一条消息被放入队列时，我希望每个等待的线程都获得自己的消息副本。
我查看了内置的Queue 类，但我认为这不合适，因为使用消息似乎涉及将它们从队列中删除，因此只有 1 个客户端线程会看到每个消息。
这似乎应该是一个常见的用例，有人可以推荐一个解决方案吗？
【问题讨论】：
标签： python multithreading queue message-queue
我认为解决此问题的典型方法是为每个线程使用单独的消息队列，并将消息推送到先前已注册有兴趣接收此类消息的每个队列。
这样的东西应该可以工作，但它是未经测试的代码......
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()
【讨论】：
multiprocessing 模块提供了更好的支持，但这是针对子进程而不是线程的。我猜这是因为进程间通信通常比线程间通信更复杂，因为线程自然共享同一个堆。
我认为这是一个更直接的示例（取自 Python Lib 中的队列示例）
from threading import Thread
from Queue import Queue
num_worker_threads = 2
def worker():
while True:
item = q.get()
do_work(item)
q.task_done()
q = Queue()
for i in range(num_worker_threads):
t = Thread(target=worker)
t.daemon = True
t.start()
for item in source():
q.put(item)
q.join() # block until all tasks are done
【讨论】：