Java - Warten auf mehrere BlockingQueue's parallel

Question

Ich habe zwei getrennte blockierende Warteschlangen. Die Clients verwenden normalerweise entweder die erste der zweiten blockierenden Warteschlangen, um die zu verarbeitenden Elemente abzurufen.

In einigen Fällen sind die Clients an Elementen aus den beiden blockierenden Warteschlangen interessiert, je nachdem, welche Warteschlange die Daten zuerst bereitstellt.

Wie kann ein Client parallel auf die beiden Warteschlangen warten?

Answer 1

Sie könnten versuchen, poll() in einer Art von Schleife zu verwenden, um nur eine bestimmte Zeitspanne für eine Warteschlange zu warten, bevor Sie die andere abfragen.

Ansonsten würde ich sagen, die blockierenden Operationen für jede Warteschlange in separaten Threads ausführen und eine Callback-Schnittstelle zu Ihrer Hauptanwendung ist eine andere, etwas komplexere Option.

Answer 2

würde ich eine Art von aggregierten Warteschlange erstellen, die Operationen auf reale Warteschlangen übertragen werden:

class AggregatedQueue<E> /*implements Queue<E>*/ { 
    final List<BlockingQueue<E>> queues; 
    final ExecutorService service; 

    final AtomicBoolean shutdown = new AtomicBoolean(false); 
    final BlockingQueue<E> aggregatedQueue = new LinkedBlockingQueue<>(); 

    public AggregatedQueue(List<BlockingQueue<E>> queues) { 
     this.queues = queues; 
     this.service = Executors.newFixedThreadPool(queues.size()); 
    } 

    public void start() { 
     for (BlockingQueue<E> queue : queues) { 
      service.submit((Runnable)() -> { 
       while (!shutdown.get()) { 
        try { 
         aggregatedQueue.add(queue.take()); 
        } catch (InterruptedException e) { 
         // handle 
        } 
       } 
      }); 
     } 
    } 

    //@Override 
    public E take() throws InterruptedException { 
     return aggregatedQueue.take(); 
    } 
} 

Um es zu nutzen, sollten Sie Ihre Warteschlangen im Konstruktor übergeben und start Methode aufrufen. Es würde 1 Thread für jede Warteschlange empfangsbereit sein und das übergebene Element an die aggregierte Warteschlange übergeben.

Answer 3

Ich stieß auf das gleiche Problem, und endete damit meine eigene gleichzeitige Warteschlange zu schreiben, um dieses Nutzungsmuster zu unterstützen.

Da das Java-Blocking primitived es nicht erlaubt, mehr als ein Objekt zu blockieren, wurde die Lösung an die Sammlung selbst übergeben: Linked Blocking Multi Queue ist eigentlich ein Set von Warteschlangen, die durch die Köpfe verbunden sind. Elemente werden einzelnen "Unterwarteschlangen" angeboten, aber von einem einzigen Ort abgerufen (was blockierende Operationen unterstützt).