java - ejemplo - ¿Cómo bloquear hasta que un BlockingQueue esté vacío?

blockingqueue java ejemplo

Question

Estoy buscando una forma de bloquear hasta que un BlockingQueue esté vacío.

Sé que, en un entorno multiproceso, mientras haya productores que coloquen elementos en BlockingQueue , puede haber situaciones en las que la cola se vacíe y unos nanosegundos después está llena de elementos.

Pero, si solo hay un productor, entonces es posible que desee esperar (y bloquear) hasta que la cola esté vacía después de que haya dejado de colocar elementos en la cola.

Java / Pseudocódigo:

// Producer code
BlockingQueue queue = new BlockingQueue();

while (having some tasks to do) {
    queue.put(task);
}

queue.waitUntilEmpty(); // <-- how to do this?

print("Done");

¿Tiene alguna idea?

EDITAR : Sé que envolver BlockingQueue y usar una condición adicional sería suficiente, solo estoy preguntando si hay algunas soluciones prefabricadas y / o mejores alternativas.

Answer 1

No es exactamente lo que quiere hacer, pero usar un SynchronousQueue tendría un efecto muy similar al de su Java / Pseudocódigo, es decir, el productor bloqueará hasta que todos los datos hayan sido recuperados por algún consumidor.

La única diferencia es el bloqueo del productor en cada lanzamiento hasta que el consumidor llega a recuperar los datos, en lugar de solo una vez al final. No estoy seguro de si eso marcaría una diferencia en su caso. Yo esperaría que solo haga una diferencia notable, si la tarea realizada por el productor es algo costosa.

Answer 2

Su caso de uso debe ser bastante especial porque lo más normal es que solo desee bloquear al productor cuando la cola está llena, pero no esperar hasta que esté vacía.

De todos modos, esto es factible. Creo que girar hasta que sea isEmpty devuelve verdadero no es isEmpty ineficaz porque el productor girará localmente, es decir, accederá a su propio caché, no golpeará el bus. Sin embargo, consumirá tiempo de CPU ya que el hilo permanece programable. Pero la rotación local es definitivamente la manera más fácil. De lo contrario, veo dos opciones:

1. Usando wait + notify como @niculare sugerido
2. De alguna forma, haga que el primer consumidor que advierta la cola esté vacío para notificar al productor de una manera libre de bloqueos; esto será más lento pero degradará "más" con gracia

Answer 3

Una solución simple usando wait() y notify() :

// Producer:
synchronized(queue) {
    while (!queue.isEmpty())
        queue.wait(); //wait for the queue to become empty
    queue.put();
}

//Consumer:
synchronized(queue) {
    queue.get();
    if (queue.isEmpty())
        queue.notify(); // notify the producer
}

Answer 4

Entiendo que ya podría haber un montón de hilos que sondearon activamente o tomaron la cola, pero todavía no me siento bien con respecto a su flujo / diseño.

La cola se vuelve vacía no significa que las tareas agregadas previamente hayan finalizado, algunos de los elementos podrían tardar años en procesarse, por lo que no es demasiado útil verificar si está vacío.

Entonces, lo que debes hacer es olvidarte de BlockingQueue , puedes utilizarlo como cualquier otra colección. Traduzca los artículos en Collections de Callable y haga uso de ExecutorService.invokeAll() .

    Collection<Item> queue = ...
    Collection<Callable<Result>> tasks = new ArrayList<Callable<Result>>();

    for (Item item : queue) {
        tasks.add(new Callable<Result>() {

            @Override
            public Result call() throws Exception {
                // process the item ...

                return result;
            }
        });
    }

    // look at the results, add timeout for invokeAll if necessary
    List<Future<Result>> results = executorService.invokeAll(tasks);

    // done

Este enfoque le dará un control total de cuánto tiempo puede esperar su productor y el manejo adecuado de las excepciones.

Answer 5

Según la documentación , BlockingQueue se puede bloquear en "Eliminar" por métodos: take() o poll(time, unit) lugar de poll()