Потоки последовательного и параллельного выполнения потоков Java8 дают разные результаты?

Запуск следующего примера потока в Java8:

    System.out.println(Stream
        .of("a", "b", "c", "d", "e", "f")
        .reduce("", (s1, s2) -> s1 + "/" + s2)
    );

дает:

/a/b/c/d/e/f

Что, конечно, не удивительно. Из-за http://docs.oracle.com/javase/8/docs/api/index.html?overview-summary.html не имеет значения, выполняется ли поток последовательно или параллельно:

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

AFAIK reduce() является детерминированным, а (s1, s2) -> s1 + "/" + s2 ассоциативным, поэтому добавление parallel() должно давать тот же результат:

    System.out.println(Stream
            .of("a", "b", "c", "d", "e", "f")
            .parallel()
            .reduce("", (s1, s2) -> s1 + "/" + s2)
    );

Однако результат на моей машине:

/a//b//c//d//e//f

Что здесь не так?

BTW: использование (предпочтительный) .collect(Collectors.joining("/")) вместо reduce(...) дает тот же результат a/b/c/d/e/f для последовательного и параллельного выполнения.

Подробности JVM:

java.specification.version: 1.8
java.version: 1.8.0_31
java.vm.version: 25.31-b07
java.runtime.version: 1.8.0_31-b13

Ответы

Ответ 1

Из уменьшенной документации:

Значение идентификатора должно быть идентификатором для функции аккумулятора. Это означает, что для всех t, accumulator.apply(identity, t) равен t.

Что не так в вашем случае - "и" a "создает" /a ".

Я извлек функцию аккумулятора и добавил распечатку, чтобы показать, что происходит:

BinaryOperator<String> accumulator = (s1, s2) -> {
    System.out.println("joining \"" + s1 + "\" and \"" + s2 + "\"");
    return s1 + "/" + s2;
};
System.out.println(Stream
                .of("a", "b", "c", "d", "e", "f")
                .parallel()
                .reduce("", accumulator)
);

Это пример вывода (он отличается между прогонами):

joining "" and "d"
joining "" and "f"
joining "" and "b"
joining "" and "a"
joining "" and "c"
joining "" and "e"
joining "/b" and "/c"
joining "/e" and "/f"
joining "/a" and "/b//c"
joining "/d" and "/e//f"
joining "/a//b//c" and "/d//e//f"
/a//b//c//d//e//f

Вы можете добавить оператор if в свою функцию для обработки пустой строки отдельно:

System.out.println(Stream
        .of("a", "b", "c", "d", "e", "f")
        .parallel()
        .reduce((s1, s2) -> s1.isEmpty()? s2 : s1 + "/" + s2)
);

Как заметил Марк Тополник, проверка s2 не требуется, так как аккумулятор не должен быть коммутативной функцией.

Ответ 2

Чтобы добавить к другому ответу,

Возможно, вы захотите использовать Mutable reduction, документ укажет, что делает что-то вроде

String concatenated = strings.reduce("", String::concat)

Дает плохую производительность.

Мы получим желаемый результат, и он будет работать параллельно. Однако мы, возможно, не очень довольны производительностью! Такой реализация будет выполнять большое количество операций копирования строк, а запуск время будет O (n ^ 2) в количестве символов. Более результативный подходом было бы накопление результатов в StringBuilder, который является изменяемым контейнером для накопления строк. Мы можем использовать та же техника для параллелизирования изменчивого сокращения, как и обычные сокращение.

Итак, вместо этого вы должны использовать StringBuilder.

Ответ 3

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

Все это под документацией сокращения, которая гласит:

Значение идентификатора ДОЛЖНО быть идентификатором для функции аккумулятора. Это означает, что для всех t, accumulator.apply(identity, t) равен t.

Мы можем легко доказать, что путь кода, ассоциативность нарушена:

static private void isAssociative() {
     BinaryOperator<String> operator = (s1, s2) -> s1 + "/" + s2;
     String result = operator.apply("", "a");
     System.out.println(result); 
     System.out.println(result.equals("a")); 
}

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