progreso - progressbar java eclipse

Mostrar el progreso del procesamiento de flujo de Java 8

Question

Tengo un Stream procesa unos cuantos millones de elementos. El algoritmo Map-Reduce detrás de él toma unos pocos milisegundos, por lo que la tarea se completa en veinte minutos.

Stream<MyData> myStream = readData();
MyResult result = myStream
    .map(row -> process(row))
    .peek(stat -> System.out.println("Hi, I processed another item"))
    .reduce(MyStat::aggregate);

Me gustaría una forma de mostrar el progreso general, en lugar de imprimir una línea por elemento (lo que da como resultado miles de líneas por segundo, requiere tiempo y no proporciona ninguna información útil sobre el progreso general). Me gustaría mostrar algo similar a:

 5% (08s)
10% (14s)
15% (20s)
...

¿Cuál sería la mejor manera (y / o la más fácil) de hacer esto?

Answer 1

Como han señalado otros: Esto tiene algunas advertencias. En primer lugar, se supone que las corrientes no deben usarse para algo como esto.

En un nivel más técnico, uno podría argumentar más:

• Una corriente puede ser infinita.
• Incluso si conoce el número de elementos: este número puede estar distorsionado por operaciones como filter o flatMap
• Para un flujo paralelo, el seguimiento del progreso impondrá un punto de sincronización
• Si hay una operación de terminal que es costosa (como la agregación en su caso), entonces el progreso informado podría no reflejar sensiblemente el tiempo de cálculo

Sin embargo, teniendo esto en cuenta, un enfoque que podría ser razonable para su caso de solicitud es el siguiente:

Podría crear una Function<T,T> que se pasa a un map de la secuencia. (Al menos, preferiría usar más que peek un peek a la secuencia, como se sugiere en otra respuesta). Esta función podría realizar un seguimiento del progreso, utilizando un AtomicLong para contar los elementos. Para mantener las cosas separadas por separado, este progreso se puede reenviar a un Consumer<Long> , que se encargará de la presentación.

La "presentación" aquí se refiere a la impresión de este progreso en la consola, normalizado o como porcentajes, en referencia a un tamaño que podría conocerse dondequiera que se cree el consumidor. Pero el consumidor también puede encargarse de imprimir solo, por ejemplo, cada décimo elemento, o solo imprimir un mensaje si han transcurrido al menos 5 segundos desde el anterior.

import java.util.Iterator;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

public class StreamProgress
{
    public static void main(String[] args)
    {
        int size = 250;
        Stream<Integer> stream = readData(size);

        Consumer<Long> progressConsumer = progress -> 
        {
            // "Filter" the output here: Report only every 10th element
            if (progress % 10 == 0)
            {
                double relative = (double) progress / (size - 1);
                double percent = relative * 100;
                System.out.printf(
                    "Progress %8d, relative %2.5f, percent %3.2f/n",
                    progress, relative, percent);
            }
        };

        Integer result = stream
            .map(element -> process(element))
            .map(progressMapper(progressConsumer))
            .reduce(0, (a, b) -> a + b);

        System.out.println("result " + result);
    }

    private static <T> Function<T, T> progressMapper(
        Consumer<? super Long> progressConsumer)
    {
        AtomicLong counter = new AtomicLong(0);
        return t -> 
        {
            long n = counter.getAndIncrement();
            progressConsumer.accept(n);
            return t;
        };

    }

    private static Integer process(Integer element)
    {
        return element * 2;
    }

    private static Stream<Integer> readData(int size)
    {
        Iterator<Integer> iterator = new Iterator<Integer>()
        {
            int n = 0;
            @Override
            public Integer next()
            {
                try
                {
                    Thread.sleep(10);
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
                return n++;
            }

            @Override
            public boolean hasNext()
            {
                return n < size;
            }
        };
        return StreamSupport.stream(
            Spliterators.spliteratorUnknownSize(
                iterator, Spliterator.ORDERED), false);
    }
}

Answer 2

En primer lugar, Streams no pretende lograr este tipo de tareas (a diferencia de una estructura de datos clásica). Si ya sabe cuántos elementos procesará su transmisión, puede utilizar la siguiente opción, que es, repito, no el objetivo de las transmisiones.

Stream<MyData> myStream = readData();
final AtomicInteger loader = new AtomicInteger();
int fivePercent = elementsCount / 20;
MyResult result = myStream
    .map(row -> process(row))
    .peek(stat -> {
        if (loader.incrementAndGet() % fivePercent == 0) {
            System.out.println(loader.get() + " elements on " + elementsCount + " treated");
            System.out.println((5*(loader.get() / fivePercent)) + "%");
        }
    })
    .reduce(MyStat::aggregate);

Answer 3

La posibilidad de hacer esto depende en gran medida del tipo de source que tenga en la stream . Si tiene una colección y desea aplicar algunas operaciones en ella, puede hacerlo porque sabe cuál es el tamaño de la colección y puede mantener un recuento de los elementos procesados. Pero hay una advertencia también en este caso. Si va a hacer cálculos paralelos en el flujo, esto también se vuelve más difícil.

En los casos en los que está transmitiendo datos desde fuera de la aplicación, es muy difícil que pueda modelar el progreso ya que no sabe cuándo terminará la transmisión.