Almacenamiento en memoria intermedia de datos en la aplicación java multiproceso

Question

Tengo una aplicación multiproceso que tiene un subproceso productor y varios subprocesos de consumidor. Los datos se almacenan en una recopilación segura de subprocesos compartidos y se descargan a una base de datos cuando hay suficientes datos en el búfer.

De los javadocs -

BlockingQueue<E> 

una cola que, además, apoya las operaciones que aguardan a que la cola se convierta en no vacío al recuperar un elemento, y espera a que el espacio que estén disponibles en la cola al almacenar un elemento .

take() 

Recupera y elimina la cabeza de esta cola, a la espera de ser necesario hasta que un elemento esté disponible.

Mis preguntas -

1. ¿Hay otra colección que tiene un método E [] tomar (int n)? es decir, la cola de bloqueo espera hasta que un elemento esté disponible. Lo que quiero es que deba esperar hasta que estén disponibles 100 o 200 elementos.
2. Alternativamente, ¿hay otro método que podría utilizar para abordar el problema sin sondeo?

Answer 1

Creo que la única manera es para extender alguna implementación de BlockingQueue o crear algún tipo de método de utilidad usando take:

public <E> void take(BlockingQueue<E> queue, List<E> to, int max) 
     throws InterruptedException { 

    for (int i = 0; i < max; i++) 
     to.add(queue.take()); 
} 

Answer 2

No No estoy seguro de si hay una clase similar en la biblioteca estándar que tiene take(int n) método de tipo, pero usted debe ser capaz de envolver el defecto BlockingQueue añadir esa función sin demasiados problemas, no se ¿pensar?

Escenario alternativo sería desencadenar una acción en la que coloque elementos en la colección, donde un umbral establecido por usted desencadenaría la descarga.

Answer 3

El método drainTo no es exactamente lo que está buscando, pero ¿serviría para su propósito?

http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/BlockingQueue.html#drainTo(java.util.Collection, int)

EDITAR

Se podría aplicar un lote poco más performante bloqueo takemin utilizando una combinación de take y drainTo:

public <E> void drainTo(final BlockingQueue<E> queue, final List<E> list, final int min) throws InterruptedException 
{ 
    int drained = 0; 
    do 
    { 
    if (queue.size() > 0) 
     drained += queue.drainTo(list, min - drained); 
    else 
    { 
     list.add(queue.take()); 
     drained++; 
    } 
    } 
    while (drained < min); 
} 

Answer 4

Así que esta debería ser una cola segura para la tarea que le permite bloquear la toma de un número arbitrario de elementos. Serán bienvenidos más ojos para verificar que el código de enhebrado sea correcto.

package mybq; 

import java.util.ArrayList; 
import java.util.LinkedList; 
import java.util.List; 

public class ChunkyBlockingQueue<T> { 
    protected final LinkedList<T> q = new LinkedList<T>(); 
    protected final Object lock = new Object(); 

    public void add(T t) { 
     synchronized (lock) { 
      q.add(t); 
      lock.notifyAll(); 
     } 
    } 

    public List<T> take(int numElements) { 
     synchronized (lock) { 
      while (q.size() < numElements) { 
       try { 
        lock.wait(); 
       } catch (InterruptedException e) { 
        Thread.currentThread().interrupt(); 
       } 
      } 
      ArrayList<T> l = new ArrayList<T>(numElements); 
      l.addAll(q.subList(0, numElements)); 
      q.subList(0, numElements).clear(); 
      return l; 
     } 
    } 
}