Colas de bloqueo múltiples, consumidor único

Question

Tengo múltiples BlockingQueues que contienen mensajes para enviar. ¿Es posible tener menos consumidores que colas? No quiero repetir las colas y seguir sondeándolas (ocupado esperando) y no quiero un hilo para cada cola. En cambio, me gustaría tener un hilo que se despierte cuando un mensaje esté disponible en cualquiera de las colas.

Answer 1

Un truco que podría hacer es tener una cola de colas. Entonces, lo que harías es tener una sola cola de bloqueo a la que se suscriben todos los hilos. Luego, cuando encola algo en una de sus BlockingQueues, también pone en cola su cola de bloqueo en esta única cola. Por lo que tendría algo como:

BlockingQueue<WorkItem> producers[] = new BlockingQueue<WorkItem>[NUM_PRODUCERS]; 
BlockingQueue<BlockingQueue<WorkItem>> producerProducer = new BlockingQueue<BlockingQueue<WorkItem>>(); 

Entonces, cuando llegue un nuevo elemento de trabajo:

void addWorkItem(int queueIndex, WorkItem workItem) { 
    assert queueIndex >= 0 && queueIndex < NUM_PRODUCERS : "Pick a valid number"; 
    //Note: You may want to make the two operations a single atomic operation 
    producers[queueIndex].add(workItem); 
    producerProducer.add(producers[queueIndex]); 
} 

Ahora sus consumidores pueden todos bloque en el producerProducer. No estoy seguro de cuán valiosa sería esta estrategia, pero sí logra lo que desea.

Answer 2

El LinkedBlockingMultiQueue hace lo que está pidiendo. No permite al consumidor bloquear en BlockingQueues arbitrarias, pero es posible crear "sub colas" desde una única "multi queue" y lograr el mismo efecto. Los productores ofrecen en las sub colas, y los consumidores pueden bloquearse a sí mismos sondeando la única cola múltiple, esperando cualquier elemento.

También admite prioridades, es decir, toma elementos de algunas colas antes de considerar otras.

Ejemplo:

LinkedBlockingMultiQueue<Int, String> q = new LinkedBlockingMultiQueue<>(); 
q.addSubQueue(1 /* key */, 10 /* priority */); 
q.addSubQueue(2 /* key */, 10 /* priority */); 
LinkedBlockingMultiQueue<Int, String>.SubQueue sq1 = q.getSubQueue(1); 
LinkedBlockingMultiQueue<Int, String>.SubQueue sq2 = q.getSubQueue(2); 

A continuación, puede ofrecer y sondeo:

sq1.offer("x1"); 
q.poll(); // "x1" 
sq2.offer("x2"); 
q.poll(); // "x2" 

responsabilidad: yo soy el autor de la biblioteca.