Java8 IntStream с лимитом + параллель

Question

Я пытаюсь пройти сертификацию Oracle 1Z0-809, я просто кодирую какой-то Дамп и наконец-то попал сюда.

public final class ParallelStreams{
   private long d=0,add=0,after=0;
   public static void main(String[] args){
       final ParallelStreams clazz = new ParallelStreams();
       IntStream.iterate(1,p->p+1)
            .peek(a->clazz.add++)
            .limit(5)
            .peek(a->clazz.after++)
            .parallel()
            .forEach(i->++clazz.d);

    System.out.println("clazz = " + clazz.d+" "+clazz.add+" "+clazz.after);
    }    
}

Я знаю, что это не лучшая вещь. Но что-то действительно привлекло мое внимание. Вывод на это что-то вроде

clazz = 5 15451 5

Говорит, что 1-кратный метод был вызван изумленно 15451 раз, даже когда предел составляет всего 5? Как это следует понимать, когда все потоки пропускают этот фрагмент кода 15451 раз, чтобы окончательно остановить его, когда в потоке было 5? Или так?

Извините, если вопрос ясен, но действительно важен для меня.

Answer 1

Вы звоните limit в параллельном потоке. limit зависит от порядка встречи, и при параллельном выполнении он должен буферизовать все предыдущие элементы, чтобы определить, следует ли перемещать конкретный элемент на следующий шаг или нет. Следовательно, он буферизует больше элементов, прежде чем принять это решение. Ваш подход неизменно последовательный, и вы пытаетесь выполнить его параллельно, что очень неэффективно и не даст вам никакого ускорения. Один из способов преодолеть это - удалить .parallel() из вашего потокового конвейера следующим образом:

   IntStream.iterate(1,p->p+1)
        .peek(a->clazz.add++)
        .limit(5)
        .peek(a->clazz.after++)
        .forEach(i->++clazz.d);

Теперь вы получите правильный результат. Другое решение состоит в том, чтобы удалить порядок столкновения, наложенный IntStream.iterate следующим образом.

IntStream.range(0, Integer.MAX_VALUE)
                .peek(a->clazz.add++)
                .limit(5)
                .peek(a->clazz.after++)
                .parallel()
                .forEach(i->++clazz.d);

Это решение может дать вам некоторое параллельное ускорение.

Answer 2

Когда я выполнил вашу программу, я получил вывод:

clazz = 5 20 5

Добавление некоторых операторов печати:

IntStream.iterate(1,p->p+1)
    .peek(a->{ System.out.println("add - " + a); clazz.add++; })
    .limit(5)
    .peek(a->{ System.out.println("after - " + a); clazz.after++; })
    .parallel()
    .forEach(i->{ System.out.println("d - " + i); ++clazz.d; });

Вывод:

add - 1025
add - 3073
add - 1
add - 6145
add - 2
add - 3074
add - 1026
add - 3075
add - 3
add - 6146
add - 6147
add - 4
add - 3076
add - 1027
add - 3077
add - 5
add - 6148
add - 6149
add - 1028
add - 1029
after - 3
after - 2
after - 5
d - 2
d - 3
d - 5
after - 1
after - 4
d - 1
d - 4
clazz = 5 20 5

Чтобы понять это, вот некоторые выдержки из документации.

java.util.stream

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

IntStream.peek

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

IntStream.limit

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

peek не знает о limit операции, поэтому она выполняется случайным образом N раз до выполнения limit. Обратите внимание, что после печати строк с after - и d - строка с add - не печатается.

Эти строки after - и d - печатают от 1 до 5, но ни peek, ни forEach не гарантируют порядок потока при работе с параллельными потоками.