Показывать ход обработки потока Java 8

Ответ 1

Прежде всего, Streams не предназначены для достижения таких задач (в отличие от классической структуры данных). Если вы уже знаете, сколько элементов будет обрабатывать ваш поток, вы можете пойти со следующей опцией, которая, я повторяю, не целью потоков.

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

Ответ 2

Как уже отмечали другие: здесь есть некоторые оговорки. Прежде всего, потоки не должны использоваться для чего-то подобного.

На более техническом уровне можно утверждать:

• Поток может быть бесконечным
• Даже если вы знаете количество элементов: это число может быть искажено такими операциями, как filter или flatMap
• Для параллельного потока отслеживание хода выполнения обеспечит точку синхронизации
• Если есть терминальная операция, которая стоит дорого (например, агрегация в вашем случае), то сообщаемый прогресс может даже не отражать время вычисления

Однако, помня об этом, один подход, который может быть разумным для вашего случая применения, заключается в следующем:

Вы можете создать Function<T,T> которая передается на map потока. (По крайней мере, я бы предпочел это использовать peek on the stream, как предложено в другом ответе). Эта функция может отслеживать прогресс, используя AtomicLong для подсчета элементов. Чтобы отделить отдельные вещи, этот прогресс можно было бы просто отправить Consumer<Long>, который позаботится о презентации.

"Представление" здесь относится к печати этого прогресса на консоли, в нормализованном виде или в процентах, ссылаясь на размер, который может быть известен везде, где создается потребитель. Но потребитель может также позаботиться только о печати, например, каждого 10-го элемента, или распечатать сообщение, только если прошло не менее 5 секунд с момента предыдущего.

import java.util.Iterator;
import java.util.Locale;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import java.util.function.LongConsumer;
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);

        LongConsumer 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(Locale.ENGLISH,
                    "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(
        LongConsumer 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);
    }
}

Ответ 3

Возможность этого очень зависит от типа source вас есть в stream. Если у вас есть коллекция, и вы хотите применить к ней некоторые операции, вы можете сделать это, потому что знаете, что такое размер коллекции, и вы можете хранить количество обработанных элементов. Но в этом случае есть оговорка. Если вы будете выполнять параллельные вычисления в потоке, это также становится более сложным.

В тех случаях, когда вы передаете данные извне приложения, очень сложно, что вы можете моделировать прогресс, поскольку вы не знаете, когда поток закончится.