Code di blocco multiple, utente singolo

Question

Ho più BlockingQueues contenenti messaggi da inviare. È possibile avere meno consumatori delle code? Non voglio fare il giro delle code e continuare a interrogarle (occupato in attesa) e non voglio un thread per ogni coda. Invece, mi piacerebbe avere un thread che si sveglia quando un messaggio è disponibile su una qualsiasi delle code.

Answer 1

Un trucco che si potrebbe fare è avere una coda di code. Quindi quello che dovresti fare è avere un'unica coda di blocco a cui tutti i thread si iscrivono. Quindi, quando accodati qualcosa in uno dei tuoi BlockingQueues, accoderai anche la tua coda di blocco su questa singola coda. Così si avrebbe qualcosa come:

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

Poi, quando si ottiene un nuovo elemento di lavoro:

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]); 
} 

Ora i vostri consumatori possono tutti blocco sul producerProducer. Non sono sicuro di quanto sia preziosa questa strategia, ma realizza ciò che desideri.

Answer 2

Il LinkedBlockingMultiQueue fa quello che stai chiedendo. Non consente al consumatore di bloccare arbitrariamente BlockingQueues, ma è possibile creare "sottocodici" da un'unica "coda multipla" e ottenere lo stesso effetto. I produttori offrono nelle code secondarie e i consumatori possono bloccarsi da soli eseguendo il polling della singola coda multi, in attesa di qualsiasi elemento.

Supporta anche le priorità, ovvero prendendo elementi da alcune code prima di considerarne altri.

Esempio:

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

Quindi è possibile offrire e sondaggio:

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

Disclaimer: io sono l'autore della biblioteca.