Параллельный поток ведет себя по-разному к потоку

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

    List<String> list = Arrays.asList("1", "2", "3");
    String resultParallel = list.parallelStream().collect(StringBuilder::new,
            (response, element) -> response.append(" ").append(element),
            (response1, response2) -> response1.append(",").append(response2.toString()))
            .toString();
    System.out.println("ResultParallel: " + resultParallel);

    String result = list.stream().collect(StringBuilder::new,
            (response, element) -> response.append(" ").append(element),
            (response1, response2) -> response1.append(",").append(response2.toString()))
            .toString();

    System.out.println("Result: " + result);

Результат Параллельный: 1, 2, 3

Результат: 1 2 3

Может кто-нибудь объяснить, почему это происходит, и как я получаю непараллельную версию, чтобы дать тот же результат, что и параллельная версия?

Ответы

Ответ 1

В качестве примечания, даже если заменить , с пробелом в combiner, ваши результаты по - прежнему будут отличаться (слегка изменил код, чтобы сделать его более удобным для чтения):

String resultParallel = list.parallelStream().collect(
            StringBuilder::new,
            (builder, elem) -> builder.append(" ").append(elem),
            (left, right) -> left.append(" ").append(right)).toString();

    String result = list.stream().collect(
            StringBuilder::new,
            (builder, elem) -> builder.append(" ").append(elem),
            (left, right) -> left.append(" ").append(right)).toString();


  System.out.println("ResultParallel: ->" + resultParallel + "<-"); // -> 1  2  3  4<-
  System.out.println("Result: ->" + result + "<-"); // -> 1 2 3 4<-

Обратите внимание, что у вас слишком много пробелов.

В java-doc есть подсказка:

объединитель... должен быть совместим с функцией аккумулятора

Если вы хотите присоединиться, есть более простые варианты:

String.join(",", yourList)
yourList.stream().collect(Collectors.joining(","))

Ответ 2

Метод Java 8 Stream.collect имеет следующую подпись:

<R> R collect(Supplier<R> supplier,
              BiConsumer<R, ? super T> accumulator,
              BiConsumer<R, R> combiner);

Где BiConsumer<R, R> combiner вызывается только в параллельных потоках (чтобы объединить частичные результаты в один контейнер), поэтому вывод вашего первого фрагмента кода:

ResultParallel: 1, 2, 3

В sequential версии combiner не вызывается (см. Этот ответ), поэтому следующий оператор игнорируется:

(response1, response2) -> response1.append(",").append(response2.toString())

и результат отличается:

1 2 3

Как это исправить? Проверьте ответ @Eugene или этот вопрос и ответы.

Ответ 3

Чтобы понять, почему это происходит неправильно, подумайте об этом из javadoc.

accumulator - ассоциативная, неинтерферирующая функция без сохранения состояния, которая должна сворачивать элемент в контейнер результата.

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

Это говорит о том, что не имеет значения, собираются ли элементы путем "накапливания" или "объединения" или некоторой комбинации этих двух. Но в вашем коде аккумулятор и объединитель объединяются с использованием другого разделителя. Они не "совместимы" в том смысле, который требуется для javadoc.

Это приводит к несогласованным результатам в зависимости от того, используются ли последовательные или параллельные потоки.

В параллельном случае поток разделяется на субпотоки ^1, которые обрабатываются разными потоками. Это приводит к отдельной коллекции для каждого подпотока. Коллекции затем объединяются.
В последовательном случае поток не расщепляется. Вместо этого поток просто накапливается в одну коллекцию, и объединение не должно происходить.

Замечания:

В общем случае для потока такого размера, выполняющего простое преобразование, parallelStream() может замедлить работу.
В этом конкретном случае узким местом с версией parallelStream() будет этап объединения. Это последовательный шаг, и он выполняет то же количество копирования, что и весь последовательный конвейер. Таким образом, на самом деле распараллеливание, безусловно, будет замедлять работу.
На самом деле, лямбды не ведут себя правильно. Они добавляют дополнительное пространство в начале и удваивают пробелы, если используется combiner. Более правильная версия:
```
String result = list.stream().collect(
    StringBuilder::new,
    (b, e) -> b.append(b.isEmpty() ? "" : " ").append(e),
    (l, r) -> l.append(l.isEmpty() ? "" : " ").append(r)).toString();
```
Класс Joiner - это гораздо более простой и эффективный способ конкатенации потоков. (Кредит: @Евгений)

^{1 - В этом случае подпотоки имеют только один элемент.}^{Для более длинного списка вы обычно получаете столько подпотоков, сколько есть рабочих потоков, а подпотоки будут содержать несколько элементов.}