Potrzebuję asynchronicznej kolejki komunikatów bezpiecznej dla wątków

Question

Szukam klasy Python (najlepiej część standardowego języka, a nie biblioteki trzeciej), aby zarządzać asynchronicznym przesyłaniem wiadomości w stylu rozgłaszania.

Będę miał jeden wątek, który umieszcza komunikaty w kolejce (metoda "putMessageOnQueue" nie może blokować), a następnie wiele innych wątków, które będą wszystkie czekały na wiadomości, przypuszczalnie nazywając niektóre blokujące funkcję "waitForMessage". Gdy wiadomość jest umieszczona w kolejce, chcę, aby każdy z oczekujących wątków otrzymał własną kopię wiadomości.

Spojrzałem na wbudowaną klasę Queue, ale nie sądzę, że jest to odpowiednie, ponieważ zużywanie wiadomości wydaje się wymagać usunięcia ich z kolejki, więc tylko jeden wątek klienta zobaczyłby każdy z nich.

Wygląda na to, że powinien to być typowy przypadek, czy ktoś może polecić rozwiązanie?

Answer 1

Myślę, że typowym podejściem do tego jest użycie osobnej kolejki komunikatów dla każdego wątku i wepchnięcie komunikatu do każdej kolejki, która wcześniej zarejestrowała zainteresowanie otrzymywaniem takich wiadomości.

Coś jak to powinno działać, ale to jest kod niesprawdzone ...

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

Answer 2

myślę, że to jest bardziej prosty przykład do przodu (przykład wzięty z kolejki w 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