Параллельная обработка с бесконечным потоком в Java

Ответ 1

Реальная причина в том, что упорядоченная параллель .distinct() - это полная барьерная операция, описанная в документации:

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

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

Как и для решения, @user140547 прав: добавьте .unordered() до .distinct(), это переключает алгоритм distinct() на неупорядоченный (который просто использует общий ConcurrentHashMap для хранения всех наблюдаемых элементов и испускает каждый новый элемент в вниз по течению). Обратите внимание, что добавление .unordered() после .distinct() не поможет.

Ответ 2

Stream.iterate возвращает "бесконечный последовательный упорядоченный поток". Поэтому параллельный параллельный поток не слишком полезен.

В соответствии с описанием Пакет потока:

Для параллельных потоков ослабление ограничения порядка может иногда обеспечивать более эффективное выполнение. Некоторые агрегированные операции, такие как фильтрация дубликатов (distinct()) или сгруппированные сокращения (Collectors.groupingBy()), могут быть реализованы более эффективно, если упорядочение элементов не имеет значения. Точно так же операции, которые по своей сути связаны с порядком, например limit(), могут требовать буферизации для обеспечения правильного упорядочения, подрывая преимущество parallelism. В тех случаях, когда поток имеет порядок встреч, но пользователь не особо заботится об этом порядке, явно отключая поток с неупорядоченным(), может улучшить параллельную производительность для некоторых операций с состоянием или терминалом. Однако большинство потоковых конвейеров, таких как пример "суммы веса блоков" выше, все еще эффективно распараллеливаются даже при ограничениях порядка.

это, по-видимому, имеет место в вашем случае, используя unordered(), он печатает 0,1.

    IntStream.iterate(0, i -> (i + 1) % 2)
            .parallel()
            .unordered()
            .distinct()
            .limit(10)
            .forEach(System.out::println);

Ответ 3

Я знаю, что код неверен и, как это предлагается в решении, если мы переместим предел до выделения, у нас не будет бесконечного цикла.

Параллельная функция использует концепцию fork и join для распределения работы, которая выделяет весь доступный поток для работы, а не отдельный поток.

Мы справедливо ожидаем бесконечный цикл, так как многократный поток бесконечно обрабатывает данные и ничего не останавливает их, так как предел 10 никогда не попадает после различий.

Возможно, возможно, что он продолжает пытаться развиваться и никогда не пытается присоединиться, чтобы продвинуть его вперед. Но все же я считаю его дефектом в java более чем что-либо еще.

Ответ 4

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

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

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

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

Без ограничений я подозреваю, что для обработки бесконечных потоков используется другой код: вместо того, чтобы ждать заполнения 10, результаты сообщаются как обнаруженные. Его немного похоже на создание итератора, который сообщает hasNext() = true, сначала производит 0, затем 1, затем следующий() вызов вешает вечно без создания результата - в параллельном случае что-то ждет несколько отчетов, чтобы он мог правильно комбинировать/заказывайте их перед выводом, в то время как в серийном случае он делает то, что он может затем висит.

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