Сортировка и отчетность сразу обрабатывают поток?

Ответ 1

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

Промежуточные операции с состоянием должны хранить данные, в некоторых случаях вплоть до момента хранения всех элементов, прежде чем они смогут передать элемент вниз по течению, но это не изменяет того факта, что эта работа отложена до начала операции терминала.

Его также неверно утверждать, что он должен "пересекать поток дважды". Происходят совершенно разные обходы, например, в случае sorted(), во-первых, обход заполнения источника во внутреннем буфере, который будет отсортирован, во-вторых, обход буфера. В случае distinct(), в последовательной обработке не происходит второго обхода, внутренний HashSet используется только для определения того, следует ли передавать элемент вниз по течению.

Поэтому, когда вы запускаете

Stream<Integer> stream = Stream.of(2,1,3,5,3)
    .peek(i -> System.out.println("source: "+i))
    .distinct()
    .peek(i -> System.out.println("distinct: "+i))
    .sorted()
    .peek(i -> System.out.println("sorted: "+i));
System.out.println("commencing terminal operation");
stream.forEachOrdered(i -> System.out.println("terminal: "+i));

он печатает

commencing terminal operation
source: 2
distinct: 2
source: 1
distinct: 1
source: 3
distinct: 3
source: 5
distinct: 5
source: 3
sorted: 1
terminal: 1
sorted: 2
terminal: 2
sorted: 3
terminal: 3
sorted: 5
terminal: 5

показывая, что ничего не происходит до начала операции терминала и что элементы из источника немедленно передают операцию distinct() (если только не дублируются), тогда как все элементы буферизуются в операции sorted() перед передачей вниз по течению.

Далее можно показать, что distinct() не должен пересекать весь поток:

Stream.of(2,1,1,3,5,6,7,9,2,1,3,5,11,10)
    .peek(i -> System.out.println("source: "+i))
    .distinct()
    .peek(i -> System.out.println("distinct: "+i))
    .filter(i -> i>2)
    .findFirst().ifPresent(i -> System.out.println("found: "+i));

печать

source: 2
distinct: 2
source: 1
distinct: 1
source: 1
source: 3
distinct: 3
found: 3

Как объяснено и продемонстрировано ответом Хосе Да Силваса, количество буферизации может изменяться с упорядоченными параллельными потоками, поскольку частичные результаты должны быть скорректированы, прежде чем они смогут перейти к операциям нижестоящего уровня.

Поскольку эти операции не выполняются до того, как известна фактическая операция терминала, существует больше оптимизаций, чем в настоящее время в OpenJDK (но может произойти в разных реализациях или будущих версиях). Например, sorted().toArray() может использовать и возвращать тот же массив или sorted().findFirst() может превратиться в min() и т.д.

Ответ 2

В соответствии с javadoc оба различных и отсортированных метода представляют собой промежуточные операции с сохранением состояния.

StreamOps говорит следующее об этих операциях:

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

Но сбор потока происходит только в терминальной операции (например, toArray, collect или forEach), обе операции обрабатываются в конвейере, и данные проходят через него. Тем не менее, важно отметить порядок выполнения этих операций, javadoc метода distinct() говорит:

Для упорядоченных потоков выбор отдельных элементов является стабильным (для дублированных элементов элемент, появляющийся первым в порядке встречи, сохраняется.) Для неупорядоченных потоков не гарантируются гарантии стабильности.

Для последовательных потоков, когда этот поток сортируется, единственный элемент, проверили предыдущий, когда не сортируется HashSet используется внутри вместо этого, по этой причине, исполняющей distinct после sort результатов в более высокой производительности.

(примечание: как заметил Юджин, выигрыш в производительности может быть крошечным в этих секвенциальных потоках, особенно когда код горячий, но все же избегает создания этого дополнительного временного HashSet)

Здесь вы можете увидеть больше о порядке distinct и sort:

Потоки Java: как сделать эффективную "отличную и сортировку"?

С другой стороны, для параллельных потоков док говорит:

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

Полное барьерное действие означает, что:

Все восходящие операции должны быть выполнены до того, как начнется нисходящий поток. В Stream API есть только две полные барьерные операции:.sorted() (каждый раз) и.distinct() (в упорядоченном параллельном случае).

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

Используя противоположный порядок, сначала сортировка (неупорядоченный параллельный поток), а затем использование отдельных, помещает барьер в обоих, сначала все элементы должны обрабатываться (поток) для sort, а затем все для distinct.

Вот пример:

Function<String, IntConsumer> process = name ->
        idx -> {
            TimeUnit.SECONDS.sleep(ThreadLocalRandom
                    .current().nextInt(3)); // handle exception or use 
                                            // LockSupport.parkNanos(..) sugested by Holger
            System.out.println(name + idx);
        };

Функция ниже получает имя и перенастраивает пользователя int, который спит от 0 до 2 секунд, а затем печатает.

IntStream.range(0, 8).parallel() // n > number of cores
        .unordered() // range generates ordered stream (not sorted)
        .peek(process.apply("B"))
        .distinct().peek(process.apply("D"))
        .sorted().peek(process.apply("S"))
        .toArray(); // terminal operation

Это будет печатать, смешивать B и D, а затем все S (без барьера в distinct).

Если изменить порядок sorted и distinct:

        // ... rest
        .sorted().peek(process.apply("S"))
        .distinct().peek(process.apply("D"))
        // ... rest

Это будет печатать, все B, а затем все S, а затем все D (барьер в distinct).

Если вы хотите попробовать еще больше добавить unordered после sorted раз:

        // ... rest
        .sorted().unordered().peek(process.apply("S"))
        .distinct().peek(process.apply("D"))
        // ... rest

Это будет печатать, все B, а затем смесь S и D (без барьера в distinct снова).

Редактировать:

Немного изменил код для лучшего объяснения и использования ThreadLocalRandom.current().nextInt(3) как sugested.