BlockingQueue - blocked drainTo() methods

Question

BlockingQueue ha il metodo chiamato drainTo() ma non è bloccato. Ho bisogno di una coda che voglio bloccare ma anche in grado di recuperare oggetti in coda in un unico metodo.

Object first = blockingQueue.take(); 

if (blockingQueue.size() > 0) 
    blockingQueue.drainTo(list); 

Immagino che il codice sopra funzionerà ma sto cercando una soluzione elegante.

Answer 1

Si riferisce alla commentare nel JavaDoc:

Inoltre, il comportamento di questa operazione non è definito se la raccolta specificata viene modificata mentre l'operazione è in corso.

Credo che questo si riferisce alla raccolta list nel tuo esempio:

blockingQueue.drainTo(list); 

il che significa che non è possibile modificare list allo stesso tempo si è scaricata da blockingQueue in list. Tuttavia, la coda di blocco si sincronizza internamente in modo che quando viene chiamato il numero drainTo, e (vedere la nota in basso) venga bloccato. Se non lo facesse, non sarebbe veramente sicuro per i thread. È possibile controllare il codice sorgente e verificare che drainTo sia sicuro per i thread per quanto riguarda la coda di blocco stessa.

In alternativa, si intende che quando si chiama drainTo che si desidera bloccare fino a quando almeno un oggetto è stato aggiunto alla coda? In questo caso, si ha poca scelta diversa:

list.add(blockingQueue.take()); 
blockingQueue.drainTo(list); 

per bloccare fino a quando sono stati aggiunti uno o più elementi, e poi scolare l'intera coda nella collezione list.

Nota: A partire da Java 7, per get e put viene utilizzato un blocco separato. Le operazioni di Put ora sono consentite durante uno drainTo (e una serie di altre operazioni di Take).

Answer 2

Con l'API disponibile, non penso che diventerai molto più elegante. Oltre a poter rimuovere il test delle dimensioni.

Se si desidera recuperare in modo atomico una sequenza di elementi contigui anche se un'altra operazione di rimozione coincide, non credo nemmeno che drainTo lo garantisca.

Answer 3

Codice sorgente:

596:  public int drainTo(Collection<? super E> c) { 
       //arg. check 
603:   lock.lock(); 
604:   try { 
608:    for (n = 0 ; n != count ; n++) { 
609:     c.add(items[n]); 
613:    } 
614:    if (n > 0) { 
618:     notFull.signalAll(); 
619:    } 
620:    return n; 
621:   } finally { 
622:    lock.unlock(); 
623:   } 
624:  } 

ArrayBlockingQueue è desideroso di tornare 0. BTW, potrebbe farlo prima di prendere la serratura.

Answer 4

Ho trovato utile questo schema.

List<byte[]> blobs = new ArrayList<byte[]>(); 
if (queue.drainTo(blobs, batch) == 0) { 
    blobs.add(queue.take()); 
} 

Answer 5

Se vi capita di utilizzare Google Guava, c'è un metodo ingegnoso Queues.drain().

scoli coda come BlockingQueue.drainTo(Collection, int), ma se il richiesti numElements elementi non sono disponibili, si attende loro fino al timeout specificato.